在 Flask 項目中使用 Celery（with 工廠模式 or not）

本文隸屬於《Flask Web 開發實戰》番外系列。這篇文章會介紹如何在 Flask 項目中集成 Celery。

建立 Celery 程序

第一步是建立一個 Celery 程序實例。由於 Flask 程序實例一般會命名爲 app，爲了不衝突，咱們通常會把 Celery 實例命名爲 celery 或 celery_app：

from celery import Celery

celery = Celery(__name__, broker='pyamqp://guest@localhost//')

@celery.task
def add(x, y):
    return x + y複製代碼

組織和加載配置

大多數教程，包括目前的 Flask 文檔裏都會介紹用下面的方式加載配置：

celery.conf.update(app.config)  # 這裏的 app 是 Flask 程序實例複製代碼

也就是把 Celery 配置和 Flask 配置寫在一塊兒，而後從 Flask 程序實例的配置字典裏更新配置。

但問題是，Celery 從 4.0 開始啓用新的小寫配置名，某些配置被新的名稱替換。雖然舊的大寫配置仍然支持，但若是你打算使用小寫配置名，或是打算在將來進行遷移，這裏的配置加載方式就會失效，由於 Flask 在從文件或對象導入配置時只會導入大寫形式的配置變量。

所以，我建議將 Celery 配置寫在單獨的文件裏，不要和 Flask 配置混在一塊兒。按照 Celery 文檔的示例，你能夠在當前目錄建立一個 celeryconfig.py 文件（或是其餘名字）保存配置：

broker_url = 'pyamqp://'
result_backend = 'rpc://'

task_serializer = 'json'
result_serializer = 'json'
accept_content = ['json']
timezone = 'Europe/Oslo'
enable_utc = True複製代碼

而後使用下面的方法加載配置（使用其餘模塊名，或是在其餘路徑時，記得修改傳入的字符串）：

celery.config_from_object('celeryconfig')複製代碼

若是須要在建立 Celery 實例時傳入 broker 和 backend 配置，能夠直接寫出或是從配置模塊中導入：

from celeryconfig import broker_url

celery = Celery(__name__, broker=broker_url)複製代碼

在 Flask 程序中初始化 Celery

你能夠單首創建 Celery 程序，但咱們一般會須要爲它添加 Flask 程序上下文支持，由於有時候你的 Celery 任務函數會用到依賴於 Flask 程序上下文的某些變量。

下面咱們爲 Celery 建立了一個工廠函數，在工廠函數中建立 Celery 實例，加載配置，並實現 Flask 程序上下文支持：

from flask import Flask
from celery import Celery

from celeryconfig import broker_url


def make_celery(app):
    celery = Celery(__name__, broker=broker_url)
    celery.config_from_object('celeryconfig')

    class ContextTask(celery.Task):
        def __call__(self, *args, **kwargs):
            with app.app_context():
                return self.run(*args, **kwargs)

    celery.Task = ContextTask
    return celery

app = Flask(__name__)

celery = make_celery(app)複製代碼

在定義 Celery 任務的模塊裏，好比 tasks.py，你能夠導入這個 Celery 程序實例：

from app import celery

@celery.task
def add(x, y):
    return x + y複製代碼

在使用工廠函數的 Flask 程序中初始化 Celery

當 Flask 程序也使用工廠函數建立時，咱們能夠全局建立 Celery 程序實例，而後在建立 Flask 程序實例的工廠函數裏更新 Celery 程序配置並進行上下文設置：

from flask import Flask
from celery import Celery

from celeryconfig import broker_url
 
celery = Celery(__name__, broker=broker_url)
 

def create_app():
    app = Flask(__name__)
 
    register_celery(app)
    return app
 

def register_celery(app):
    celery.config_from_object('celeryconfig')
 
    class ContextTask(celery.Task):
        def __call__(self, *args, **kwargs):
            with app.app_context():
                return self.run(*args, **kwargs)
 
    celery.Task = ContextTask複製代碼

一樣直接導入 Celery 實例並建立任務：

from app import celery

@celery.task
def add(x, y):
    return x + y複製代碼

---

這原本是一個完整的 Celery 入門教程，但由於去年的一次硬盤損壞，對應的示例程序弄丟了，暫時沒有毅力重寫一遍，因此這篇文章只抽取了其中和 Flask 相關的內容。

由於距離初稿寫做的時間已通過去半年多，Celery 的最新版本也已是 4.3.0，若是文中有什麼疏漏，或是有更好的實現方式，歡迎評論指出。