celery使用

介绍

       Celery 是一个简单、灵活且可靠的分布式任务队列系统，用于处理实时操作和调度任务。它的设计目标是简化任务的调度和执行，并且能够与各种消息代理（如 RabbitMQ、Redis）进行集成。

使用

安装celery

pip install celery

创建celery

• 'my_project':  Celery 应用名称。通常是项目的名称或一个描述性名称，用于标识这个 Celery 实例。

• broker='redis://localhost:6379/0': 这是消息代理（Broker）的 URL。Celery 使用消息代理来分发任务。这里使用 Redis 作为消息代理，localhost:6379/0 表示 Redis 运行在本地主机的 6379 端口，并使用第 0 个数据库。

• backend='redis://localhost:6379/0': 这是结果后端（Backend）的 URL。Celery 使用结果后端来存储任务的结果。这里使用 Redis 作为结果后端，配置与 Broker 相同。

• include=['my_project.tasks']: 这是一个包含任务模块的列表。Celery 会自动加载这些模块中的任务。'my_project.tasks' 指的是 my_project 包下的 tasks 模块。

• result_expires=3600: 可选配置项，用于设置任务结果的过期时间（以秒为单位）。在这个例子中，任务结果会在 3600 秒（即 1 小时）后过期并被删除。防止结果后端中的数据无限增长，从而节省存储空间。

# celery.pyfrom celery import Celeryapp = Celery('my_project',broker='redis://localhost:6379/0',backend='redis://localhost:6379/0',include=['my_project.tasks'])# 可选配置，更多配置项请参考官方文档
app.conf.update(result_expires=3600,
)if __name__ == '__main__':app.start()

创建任务

# tasks.pyfrom celery import shared_task@shared_task
def add(x, y):return x + y@shared_task
def mul(x, y):return x * y@shared_task
def xsum(numbers):return sum(numbers)

运行celery work

celery -A my_project worker --loglevel=info

调用任务

# main.pyfrom tasks import addresult = add.delay(4, 4)
print('Task result:', result.get(timeout=10))

Flask+celery

创建celery和flask

# app.pyfrom flask import Flask
from celery import Celerydef 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)TaskBase = celery.Taskclass ContextTask(TaskBase):def __call__(self, *args, **kwargs):with app.app_context():return TaskBase.__call__(self, *args, **kwargs)celery.Task = ContextTaskreturn celeryapp = Flask(__name__)
app.config.update(CELERY_BROKER_URL='redis://localhost:6379/0',CELERY_RESULT_BACKEND='redis://localhost:6379/0'
)celery = make_celery(app)@app.route('/add/<int:a>/<int:b>')
def add(a, b):result = add_together.delay(a, b)return f'Task ID: {result.id}'@celery.task
def add_together(a, b):return a + bif __name__ == '__main__':app.run(debug=True)

运行两个任务

celery -A app.celery worker --loglevel=info
#另一个终端
python app.py