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 的信息，不得不以这种方式处理。