Flask 工厂模式使用 Celery

项目结构

在原先简单的项目里使用 Celery 是比较随意简单的，但是在改进了项目结构后，加上 Celery 和 Docker 就不一样了。我的项目结构是

app.py
celery_worker.py
Dockerfile
gunicorn.conf.py
app/
  __init__.py
  db.py
  settings.py
  tasks.py
  auth/
    __init__.py
    handler.py
    models.py
  blog/
    __init__.py
    handler.py
    models.py
...

创建 Celery 实例

项目需要使用 celery.task 作为装饰器使用，就必须创建一个全局的 celery 对象。在项目根目录下的 __init__.py （即创建 create_app 的地方）创建并配置 celery。配置 broker/backend 等等都在 settings.py 中，broker 必须要在 createa_app 之前创建，所以现在的缺点就是不能配置不同的 broker

from . import settings
from celery import Celery

celery = Celery(__name__, broker=settings.CELERY_BROKER_URL)

def create_app():

    celery.conf.update(app.config)
    # ...

创建 task, worker

将详细任务写在 tesks.py 中，方法当然加上 @celery.task 装饰器。接着就是创建一个 worker，worker 进程必须有自己的 Flask 实例，才能有上下文环境去执行 task。所以创建一个单独的脚本 celery_worker.py

#!/usr/bin/env python
from manage_risk import celery, create_app

app = create_app()
app.app_context().push()

Docker 部署

到此，已经可以通过 celery worker -A celery_worker.celery --loglevel=info 启动 celery。通过 Docker 部署的话，只要在 Dockerfile 的 CMD 加上执行语句就行了，多条执行语句之间加上 &&

FROM python:3.6

# ...
CMD gunicorn app:app -c ./gunicorn.conf.py && celery worker -A celery_worker.celery --loglevel=info