現在繼續學習在集成的框架中如何使用celery.

在Flask中使用celery

在Flask中集成celery需要做到兩點：

1. 創建celery的實例對象的名字必須是flask應用程序app的名字,否則celery啟動會失敗；

2. celery必須能順利加載初始化文件。

celery在flask中初始化

由於celery進程的運行和flask進程的運行是相互獨立的，但是在框架中我們希望只使用一份配置文件，這樣可以簡化配置的工作。

from celery import Celery
from flask import Flask

app = Flask(__name__)

def make_celery(app):
    celery = Celery(app.import_name)
    celery.conf.update(app.config)
    return celery

celery = make_celery(app)
# celery的配置文件在app的setting中；
 

問題：上面的做法程序在初始化的時候可以完成celery的初始化，但是當使用工廠模式創建app的時候，celery的初始化變得困難；

from celery import Celery
from flask import Flask

celery = None

def make_celery(app):
    celery = Celery(app.import_name)
    celery.conf.update(app.config)
    return celery

def create_app(config_name)
    app = Flask(__name__)
    config_class = config_map[config_name]
    app.config.from_object(config_class)
    # 初始化celery
    global celery
    celery = make_celery(app)
 

• 問題：由於程序初始化的時候並不能創建出app對象，所以celery啟動的時候必須先在tasks中導入app對象才能完成初始化，可能導致循環導入的錯誤；

• 解決：引入Flask-Celery-Helper，幫助我們初始化celery對象；

• 安裝Flask-Celery-Helper

pip install Flask-Celery-Helper

• 將所有額外的需要初始化的對象獨立出來在一個單獨的py模塊。

# extensions.py
from flask_celery import Celery
# 創建celery的實例
celery = Celery()

# __init__.py
from extensions import celery

def create_app(config_name)
    app = Flask(__name__)
    config_class = config_map[config_name]
    app.config.from_object(config_class)
    # 對celery進行初始化操作，可以將celery的配置寫在app的配置中
    celery.init_app(app=app)

# tasks.py
from extensions import celery

@celery.task()
def add(x,y):
    return x + y
 

註意

• Flask-Celery-Helper官方目前只支持到python3.4，但樓主使用py3.6也沒有問題；Flask-Celery-Helper不支持celery4.X的版本，否則報錯，因此需要使用celery3.x的版本；

• 調用task方法

# app.py

from tasks import add
@app.route("/index")
def index():
    """一個測試的實例"""
    print(add(3+6)) # add函數也能做普通的函數使用
    add.apply_async(args=[5,7], queue=‘eegqueue‘) # 發送異步任務，指定隊列

    return "ok!"

啟動celery

啟動celery之前需要加載flask的app的配置，因此需要創建一個app對象給celery使用。

# run_celery.py
import create_app
flask_app = create_app("develop") # 創建app的同時，對celery完成了加載配置的工作
from extensions import celery # 此時的celery對象已經在上下文中完成初始化

# 找到celery實例的位置，指定worker，指定接收某個隊列的消息，如果不指定則接收所有隊列的消息
celery -A run_celery.celery worker -Q eegqueue --loglevel=info

• 參考：

• https://pypi.org/project/Flask-Celery-Helper/#description

python之celery在flask中使用