Celery背景任务

如果您的应用程序有一个长时间运行的任务,例如处理一些上载的数据或发送电子邮件,您不希望在请求期间等待它完成。相反,使用任务队列将必要的数据发送到另一个进程,该进程将在请求立即返回时在后台运行任务。

Celery是一个功能强大的任务队列,可用于简单的后台任务以及复杂的多阶段程序和日程安排。 本指南将向您展示如何使用Flask配置Celery,但假设您已经阅读了“Celery的第一步”<https://celery.readthedocs.io/en/latest/getting-started/first-steps-with- Celery文档中的celery.html>`_ guide。

安装

Celery是一个单独的python包。使用pip从pypi安装:

$ pip install celery

配置

您需要的第一件事是Celery实例,这称为芹菜应用程序。 它与Flask中的:class:`~flask.Flask`对象具有相同的目的,仅用于Celery。 由于此实例用作您希望在Celery中执行的所有操作的入口点,例如创建任务和管理工作程序,因此其他模块必须可以导入它。

例如,您可以将其放在``tasks``模块中。 虽然您可以使用Celery而无需使用Flask进行任何重新配置,但通过子类化任务并添加对Flask应用程序上下文的支持并使用Flask配置将其连接起来,它会变得更好一些。

这是将Celery与Flask正确集成所需的全部内容:

from celery import Celery

def make_celery(app):
    celery = Celery(
        app.import_name,
        backend=app.config['CELERY_RESULT_BACKEND'],
        broker=app.config['CELERY_BROKER_URL']
    )
    celery.conf.update(app.config)

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

    celery.Task = ContextTask
    return celery

该函数创建一个新的Celery对象,使用代理配置中的代理进行配置,从Flask配置更新Celery配置的其余部分,然后创建任务的子类,将任务执行包装在应用程序上下文中。

示例任务

让我们编写一个将两个数字相加并返回结果的任务。我们将Celery的代理和后端配置为使用redis,创建一个 celery 应用程序使用上面的系数,然后使用它来定义任务。::

from flask import Flask

flask_app = Flask(__name__)
flask_app.config.update(
    CELERY_BROKER_URL='redis://localhost:6379',
    CELERY_RESULT_BACKEND='redis://localhost:6379'
)
celery = make_celery(flask_app)

@celery.task()
def add_together(a, b):
    return a + b

现在可以在后台调用此任务:

result = add_together.delay(23, 42)
result.wait()  # 65

管理工人

如果你跳进去并且已经执行了上面的代码,你会很失望地知道``.wait()``永远不会真正返回。 那是因为您还需要运行Celery工作程序来接收和执行任务。:

$ celery -A your_application.celery worker

``your_application``字符串必须指向应用程序的包或模块,它们创建``celery``对象。

现在工作者正在运行,``wait``将在任务完成后返回结果。