環境：win10

celery==4.2.1         flask==1.0.2       

1、設定celery的broker（以redis為例）

    # celery配置
    CELERY_BROKER_URL = 'redis://10.100.14.131:6379/0'
    CELERY_RESULT_BACKEND = 'redis://10.100.14.131:6379/1'

2、根據設定，將其配入到celery物件中
celery = Celery(__name__, broker=configs['test'].CELERY_BROKER_URL)

3、在工廠函式中載入celery配置檔案

4、在APP中新建tasks.py檔案如下：

from flask_mail import Message

from app import celery, mail


@celery.task
def send_mail_tasks():
    import time
    time.sleep(4)
    msg = Message('Hello',
                  # sender='[email protected]',
                  recipients=['[email protected]'])
    msg.html = '<h1>Hello World</h1>'
    mail.send(msg)
    return "Success"
 

5、對tasks任務的呼叫：

@myapp.route('/mail', methods=['GET'])
def send_mail():
    tasks.send_mail_tasks.delay()
    return "send_mail success"

6、啟動celery，請確保redis已經正常啟動：
celery -A app.myapp.tasks worker --pool=solo -l info