celery 异步发送邮件

celery4.3的使用说明：

celery是一个异步的任务队列。你可以使用它在你的应用上下文之外执行任务。说白了就是：你的响应程序可能需要执行任何消耗资源的任务都可以交给任务队
列，让你的程序自由快速地响应客户端请求。

celery有三大核心组件：

• 1、celery客户端：用于发布后台作业。当与Flask一起工作的时候客户端与Flask应用一起运行
• 2、celery workers：这些是运行后台作业的进行。celery支持本地和远程的workers，因此你可以在Flask服务器上启动一个单独的worker，随后随着你的应用需求的增加而新增加更多的worker。
• 3、消息代理：客户端通过消息队列和workers进行通信，celery支持多种方式实现这些队列，最常用的就是Redis和RabbitMQ

启动celery：在项目目录下（定义任务的目录下）终端执行：celery -A proj任务文件名.celery worker -l INFO

在使用是作如下说明：
运行celery4版本会出现以下错误：

• 1、kombu.exceptions.EncodeError: Object of type ‘Message’ is not JSON serializable
  • 原因：4.0+版本默认使用json序列化，不包括自定义的实例，如，下例的发送邮件的Message对象
  • 解决方法：app.config[“CELERY_TASK_SERIALIZER”] = “pickle”，配置信息设置为pickle序列化
• 2、kombu.exceptions.ContentDisallowed: Refusing to deserialize untrusted content of type pickle (application/x-python-serialize)
  • 原因：拒绝反序列化pickle类型的不可信内容
  • 解决方法：还是修改配置项
    • app.config[“CELERY_ACCEPT_CONTENT”] = [“pickle”]
    • app.config[“CELERY_RESULT_SERIALIZER”] = “pickle”
• 其他说明见程序注释

from flask import Flask, request, render_template, session, flash, redirect, url_for
from celery import Celery
from flask_mail import Mail, Message

app = Flask(__name__)

# celery配置
app.config["CELERY_BROKER_URL"] = "redis://127.0.0.1:6379/0"  # 消息代理的连接url
app.config["CELERY_RESULT_BACKEND"] = "redis://127.0.0.1:6379/0"  # 存储状态后运行结果
app.config["CELERY_TASK_SERIALIZER"] = "pickle"  # 默认是json序列化，设置为pickle
app.config["CELERY_ACCEPT_CONTENT"] = ["pickle"]  # 设置反序列化为pickle
app.config["CELERY_RESULT_SERIALIZER"] = "pickle"


# 163邮箱配置
app.config["MAIL_SERVER"] = "smtp.163.com"  # 电子邮箱的服务器的主机名和ip地址
app.config["MAIL_PORT"] = 25  # 电子邮箱的服务器的端口
app.config["MAIL_USE_TLS"] = True  # 启用传输层安全协议
app.config["MAIL_USERNAME"] = "你的邮箱"  # 邮箱账户用户名
app.config["MAIL_PASSWORD"] = "你的密码"  # 邮件账户的密码

app.config["SECRET_KEY"] = "sdfghjklfghjk"

# 创建一个celery对象，传入应用名和消息代理的连接url
celery = Celery(app.name, broker=app.config['CELERY_BROKER_URL'])
# celery的其他任何配置可以直接使用celery.conf.update（）通过Flask的配置直接传递
celery.conf.update(app.config)

# 创建一个email对象
mail = Mail(app)


# 定义一个异步任务，这里传参是Message的实例对象，因为设置了pickle序列化，可以直接传递自定义的实例，程序出错
@celery.task
def send_async_email(msg):
    with app.app_context():
        mail.send(msg)


# 当使用json序列化时，需要在任务中创建实例
@celery.task
def send_async_email2():
    msg = Message(
        "Hello from flask",
        sender='xxx@163.com',
        recipients=["cxxxx@sina.com"]
    )
    msg.body = "This is a test email sent from a background Celery task."
    with app.app_context():
        mail.send(msg)


@app.route("/", methods=["GET", "POST"])
def index():
    if request.method == "GET":
        return render_template("celery异步邮件.html", email=session.get("email", ""))
    email = request.form.get("email")
    session['email'] = email

    # 发送邮件（使用自定义的实例）
    # msg = Message(
    #     "Hello from flask",
    #     sender='xxxx@163.com',
    #     recipients=[request.form.get("email")]
    # )
    # msg.body = "This is a test email sent from a background Celery task."
    if request.form['submit'] == 'Send':
        # 立即发送（使用自定义的实例）
        # send_async_email.delay(msg)
        # 未设置pickle序列化时，需在任务中创建实例，执行异步任务如下：
        send_async_email2.delay()
        flash(f"Sending email to {email}")
    else:
        # 一分钟后发送（使用自定义的实例）
        # send_async_email.apply_async(args=[msg], countdown=60)
        # 未设置pickle序列化时，需在任务中创建实例，执行异步任务如下：
        send_async_email2.apply_async(countdown=60)
        flash(f'An email will be sent to {email} in one minute')
    return redirect(url_for("index"))


if __name__ == '__main__':
    app.run()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83