Flask 與 Celery 在 windows 下的集成問題

Flask 與 Celery 在 windows 下的集成問題

全部的 Web 框架內部的視圖中不適合執行須要長時間運行的任務，包括 Flask 、Django 等。這類型的任務會阻塞 Web 的響應，致使用戶在等待執行結果，對用戶不友好。發送郵件通知、數據統計等任務，執行時間長，應該自動或由用戶觸發，而後在後臺執行。

後臺執行的方式有多種，多線程、多進程均可以。但要注意，通常不要直接在 Web 框架中觸發多線程任務，由於沒法肯定 web 服務器會否進行回收線程致使任務停止，不夠可靠。能夠經過獨立的服務進程，本身寫線程任務或多進程任務完成這些操做。更加可靠和方便的方法是使用 Celery 和消息隊列的方式，讓 Celery 來管理相關的任務，同時也便於監控。

Flask 的在線文檔有一個簡單的方法集成 Celery 和 Flask Celery Based Background Tasks 。這個方法在 Unix-link 系統中工做正常，在 windows 上以工廠模式工做時，會由於 windows 的進程建立機制問題工做異常，包括像 Flask-CeleryExt / Flask-Celery-Helper 都有點問題，Celery 的 worker 子進程沒法加載 Flask 的 App Context 。

這裏提供一個各個平臺均可行的方法，假設佈局以下：

1. app.py

   這個文件以工廠模式建立 Flask App

   from flask import Flask
    from flask.ext.sqlalchmey import SQLAlchemy
   
    db = SQLAlchemy()
   
    def create_app(config):
        app = Flask(__name__)
        app.config.from_object(config)
   
        db.init_app(app)
   
        # 加載 blueprint
   
        return app

2. tasks.py

   這個文件定義 celery 任務

   from celery import current_app as current_celery
    from app import db
   
    @current_celery.task()
    def say_hello(name):
        # 數據庫操做，db.session...
        return 'Hello, %s!' % name

3. config.py

   配置文件

   class config(object):
        # ...
        CELERY_BROKER_URL = 'sqla+sqlite:///celery-broker.sqlite'
        CELERY_RESULT_BACKEND = 'db+sqlite:///celery-result.sqlite'
        CELERY_ALWAYS_EAGER = True

4. manage.py

   這個文件做爲程序入口

   from flask.ext.script import Manager
    from app import create_app
    from celery import Celery, Task
    from config import config
   
    app = create_app(config)
    manager = Manager(app)
   
    class AppContextTask(Task):
        abstract = True
   
        def __call__(self, *args, **kwargs):
            if config.CELERY_ALWAYS_EAGER:
                return super(BoundTask, self).__call__(*args, **kwargs)
            else:
                with app.app_context():
                    return super(BoundTask, self).__call__(*args, **kwargs)
   
    celery = Celery(app.import_name, broker=config.CELERY_BROKER_URL),
                    set_as_current=True, task_cls= AppContextTask)
    celery.conf.update(app.config)
   
    # ... 其它
   
    if __name__ == '__main__':
        manager.run()

在調試時，可把 CELERY_ALWAYS_EAGER 設置爲 True ，那麼全部的調用會變爲實時調用，不經過 Broker 進行，所以在 AppContextTask 中，就不須要從新注入 App Context 。

這種方法在 windows 、linux 和 mac 上已測試過可行，可是要注意，Flask 的 app 和 AppContextTask 必須在同一個文件一塊兒建立，我嘗試過以 Flask 擴展中 init_app 的方式處理時，發現新生成的 Celery 對象和原始 Celery 對象徹底不一樣，沒法保留原始 Flask app 的信息，不得不以這種方式處理。