Python 中如何使用 Celery 执行异步任务？

推荐答案

-- -------------------- ---- -------
- -- ------
--- ------- ------

- ---- ------ --
---- ------ ------ ------

--- - --------------- ------------------------------------

- --------
---------
--- ------ ---
    ------ - - -

- ------
------ - ------------ --

- ------
-------------------

本题详细解读

1. 安装 Celery

首先，你需要安装 Celery 库。可以通过以下命令安装：

pip install celery

2. 创建 Celery 应用

在 Python 中，你需要创建一个 Celery 应用实例。这个实例将用于定义和调用异步任务。通常，你需要指定一个消息代理（broker），例如 RabbitMQ 或 Redis。

from celery import Celery

app = Celery('tasks', broker='pyamqp://guest@localhost//')

3. 定义异步任务

使用 @app.task 装饰器来定义一个异步任务。这个任务可以在后台执行，而不会阻塞主线程。

@app.task
def add(x, y):
    return x + y

4. 调用异步任务

通过 .delay() 方法来调用异步任务。这个方法会将任务放入消息队列中，Celery 的 worker 会从队列中取出任务并执行。

result = add.delay(4, 4)

5. 获取任务结果

你可以通过 .get() 方法来获取异步任务的结果。这个方法会阻塞当前线程，直到任务完成并返回结果。

print(result.get())

6. 运行 Celery Worker

为了执行异步任务，你需要启动 Celery worker。可以在命令行中运行以下命令：

celery -A tasks worker --loglevel=info

其中 tasks 是你的 Celery 应用模块的名称。

7. 消息代理配置

Celery 需要一个消息代理来传递任务。常用的消息代理包括 RabbitMQ 和 Redis。你可以在创建 Celery 应用时指定消息代理的 URL。

app = Celery('tasks', broker='redis://localhost:6379/0')

8. 任务结果存储

Celery 还支持将任务结果存储在后台数据库中。你可以通过配置 backend 参数来指定结果存储的位置。

app = Celery('tasks', broker='pyamqp://guest@localhost//', backend='rpc://')

9. 任务重试与错误处理

Celery 提供了任务重试机制，可以在任务失败时自动重试。你可以通过 @app.task 装饰器的 autoretry_for 参数来配置重试策略。

@app.task(autoretry_for=(Exception,), retry_kwargs={'max_retries': 3})
def add(x, y):
    return x + y

10. 任务链与任务组

Celery 还支持任务链（chain）和任务组（group）等高级功能，可以用于构建复杂的任务流程。

from celery import chain, group

# 任务链
chain(add.s(4, 4), add.s(8)).apply_async()

# 任务组
group(add.s(i, i) for i in range(10))().get()

通过这些步骤，你可以在 Python 中使用 Celery 来执行异步任务，并处理任务的结果和错误。