网站图片自动下载建设网站的群
异步邮件发送项目目录结构
异步发送邮件/ ├── image.png ├── app.py ├── celery_config.py ├── tasks.py └── templates/└── index.html
项目文件说明：
根目录 (
异步发送邮件/)image.png- 项目相关图片app.py- Flask应用主文件，包含路由和视图函数celery_config.py- Celery配置，设置消息代理和结果后端tasks.py- 定义Celery任务，包含邮件发送功能
templates目录
index.html- 邮件发送前端页面，包含表单和JavaScript
技术栈说明：
Flask - Web框架，处理HTTP请求
Celery - 分布式任务队列，处理异步邮件发送
Redis - 作为Celery的消息代理和结果后端
Flask-Mail - 邮件发送扩展
HTML/CSS/JavaScript - 前端用户界面
这个项目实现了异步邮件发送功能，用户在前端填写表单后，任务会被提交到Celery队列异步处理，前端通过轮询方式检查任务状态。
步骤
1.先写异步的配置文件
# 创建一个celery_config.py 文件
class CeleryConfig:# 使用 Redis 作为消息代理 :6379 是redis默认端口号broker_url = 'redis://localhost:6380/0'# 任务结果存储（禁用可提升性能）result_backend = 'redis://localhost:6380/1'# 定时任务配置（可选）beat_schedule = {'every-30-seconds': {'task': 'tasks.periodic_task','schedule': 30.0, # 每30秒执行},}2.写请求页面路由
from flask import Flask,render_template,jsonify,request
from tasks import celery,send_async_email
app = Flask(__name__)
# 请求页面路由
@app.route('/')
def index():return render_template('index.html')3.写请求页面的html
<!DOCTYPE html>
<html lang="zh-CN">
<head><meta charset="UTF-8"><meta name="viewport" content="width=device-width, initial-scale=1.0"><title>邮件发送</title><style>* {box-sizing: border-box;margin: 0;padding: 0;font-family: 'Segoe UI', Tahoma, Geneva, Verdana, sans-serif;}
body {background: linear-gradient(135deg, #f5f7fa 0%, #c3cfe2 100%);padding: 20px;display: flex;justify-content: center;align-items: center;min-height: 100vh;}
.container {background-color: white;border-radius: 10px;box-shadow: 0 5px 20px rgba(0, 0, 0, 0.15);padding: 30px;width: 100%;max-width: 500px;}
h1 {text-align: center;margin-bottom: 25px;color: #2c3e50;font-size: 28px;padding-bottom: 10px;border-bottom: 2px solid #eee;}
.form-group {margin-bottom: 20px;}
label {display: block;margin-bottom: 8px;font-weight: 600;color: #34495e;font-size: 16px;}
input, textarea {width: 100%;padding: 14px;border: 1px solid #ddd;border-radius: 6px;font-size: 16px;transition: all 0.3s ease;}
input:focus, textarea:focus {outline: none;border-color: #3498db;box-shadow: 0 0 0 3px rgba(52, 152, 219, 0.2);}
textarea {min-height: 180px;resize: vertical;}
button {background: linear-gradient(to right, #3498db, #2980b9);color: white;border: none;border-radius: 6px;padding: 15px;font-size: 18px;cursor: pointer;width: 100%;transition: all 0.3s ease;font-weight: 600;letter-spacing: 1px;margin-top: 10px;}
button:hover {background: linear-gradient(to right, #2980b9, #3498db);transform: translateY(-2px);box-shadow: 0 5px 15px rgba(0, 0, 0, 0.1);}
.note {text-align: center;margin-top: 20px;color: #7f8c8d;font-size: 14px;border-top: 1px solid #eee;padding-top: 15px;}
@media (max-width: 600px) {.container {padding: 20px;}
h1 {font-size: 24px;}
input, textarea {padding: 12px;}}</style>
</head>
<body><div class="container"><h1>发送邮件</h1><form id="emailForm"><div class="form-group"><label for="email">收件邮箱:</label><input type="email" id="email" name="email" required placeholder="请输入收件人邮箱地址"></div>
<div class="form-group"><label for="subject">邮件主题:</label><input type="text" id="subject" name="subject" required placeholder="请输入邮件主题"></div>
<div class="form-group"><label for="content">邮件内容:</label><textarea id="content" name="content" required placeholder="请输入邮件内容"></textarea></div>
<button type="submit">发送邮件</button></form>
<div class="note">注意：这是一个前端演示页面，实际发送功能需要后端支持。</div></div>
</body>
</html>4.写POST请求的send路由
from flask import Flask,render_template,jsonify,request
from tasks import celery,send_async_email
app = Flask(__name__)
# 请求页面路由
@app.route('/')
def index():return render_template('index.html')
# 发送邮件路由
@app.route('/send',methods=["POST"])
def send_email():# 获取请求的数据data = request.form if request.form else request.get_json()email = data['email']subject = data['subject']content = data['content']………………………………………………………………………………………………………………………………5.写异步任务文件tasks.py
from celery import Celery
from flask import Flask
from flask_mail import Mail, Message
# 初始化Celery实例
celery = Celery(__name__)
# 导入异步配置文件
celery.config_from_object('celery_config.CeleryConfig')
def create_app():app = Flask(__name__)
# 邮件服务器配置（以网易邮箱为例）app.config['MAIL_SERVER'] = 'smtp.163.com' # SMTP服务器地址app.config['MAIL_PORT'] = 465 # SSL加密端口app.config['MAIL_USE_SSL'] = True # 启用SSL加密app.config['MAIL_USERNAME'] = '19086325659@163.com' # 发件邮箱app.config['MAIL_PASSWORD'] = 'hsbbh' # 邮箱授权码（非密码）app.config['MAIL_DEFAULT_SENDER'] = '19086325659@163.com' # 默认发件人return app
flask_app = create_app()
mail = Mail(flask_app)
# 发送邮件的方法
def send_email(to, subject, content, **kwargs):msg = Message(subject, recipients=[to])msg.body = content # 纯文本内容mail.send(msg)
# max_retries限制任务最低重试3次
@celery.task(bind=True,max_retries=3)
def send_async_email(self,email,subject,content):try:with flask_app.app_context():send_email(to=email,subject=subject,content=content)return f'邮件成功发送至{email}'# 如果发送错误则重试，并且60秒重试一次，最多3次except Exception as e:self.retry(exc = e,countdown=60)6.继续补充send路由
from flask import Flask,render_template,jsonify,request
from tasks import celery,send_async_email
app = Flask(__name__)
# 请求页面路由
@app.route('/')
def index():return render_template('index.html')
# 发送邮件路由
@app.route('/send',methods=["POST"])
def send_email():# 获取请求的数据data = request.form if request.form else request.get_json()email = data['email']subject = data['subject']content = data['content']# 触发异步任务task = send_async_email.delay(email,subject,content)return jsonify({'status':'邮件任务已提交','task_id':task.id})7.写查看任务状态路由
from flask import Flask,render_template,jsonify,request
from tasks import celery,send_async_email
app = Flask(__name__)
# 请求页面路由
@app.route('/')
def index():return render_template('index.html')
# 发送邮件路由
@app.route('/send',methods=["POST"])
def send_email():# 获取请求的数据data = request.form if request.form else request.get_json()email = data['email']subject = data['subject']content = data['content']# 触发异步任务task = send_async_email.delay(email,subject,content)return jsonify({'status':'邮件任务已提交','task_id':task.id})
# 查看状态路由
@app.route('/status/<task_id>')
def task_status(task_id):# 获取任务状态task_result = celery.AsyncResult(task_id)
return jsonify({"status": task_result.status,"result": task_result.result if task_result.ready() else None})
if __name__ == '__main__':app.run(debug=True)8.写index.html页面的js
<script>const emailForm = document.querySelector('#emailForm')emailForm.addEventListener('submit', (e) => {e.preventDefault()// 获取表单数据const email = document.querySelector('#email').valueconst subject = document.querySelector('#subject').value || '默认主题'const content = document.querySelector('#content').value || '这是一封测试邮件'// 给用户提示const note = document.querySelector('.note')note.innerHTML = '邮件正在发送中……'// 发送POST请求fetch('/send',{method:'POST',headers:{'Content-Type':'application/x-www-form-urlencoded'},body:`email=${email}&subject=${subject}&content=${content}`}).then(res => res.json()).then(data => {if (data.task_id){// console.log(data.task_id)// 开始检查任务checkTaskStatus(data.task_id)}}).catch(err => {note.innerHTML = `网络错误:${err.message}`})})function checkTaskStatus(taskid){const note = document.querySelector('.note')const checkTask = () =>{fetch(`/status/${taskid}`).then(res => res.json()).then(data => {if(data.status == 'SUCCESS'){note.innerHTML = '邮件发送成功'}else if(data.status == 'FAILURE'){note.innerHTML = '邮件发送失败'}else if(data.status == 'PENDING'){note.innerHTML = '邮件发送中'setTimeout(checkTask,2000)}else if(data.status == 'RETRY'){note.innerHTML = '尝试重连中'setTimeout(checkTask,3000)}else {note.innerHTML = '处理中'setTimeout(checkTask,3000)}}).catch(err =>{note.innerHTML = '出错啦'})}checkTask()}
</script>9.启动服务并进行调试
终端1 - 启动Redis
redis-server
终端2 - 启动Celery worker
celery -A tasks.celery worker --loglevel=info
终端3 - 启动Flask应用
python app.py