FastAPI 中如何使用 Celery？

推荐答案

在 FastAPI 中使用 Celery 可以通过以下步骤实现：

1. 安装依赖： 首先，确保安装了 celery 和 fastapi 库。可以使用以下命令安装：

   pip install celery fastapi

2. 配置 Celery： 创建一个 celery.py 文件来配置 Celery 实例：

   from celery import Celery
   
   app = Celery(
       'tasks',
       broker='redis://localhost:6379/0',
       backend='redis://localhost:6379/0'
   )

3. 定义任务： 在 tasks.py 文件中定义 Celery 任务：

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

4. 在 FastAPI 中调用 Celery 任务： 在 FastAPI 应用中调用 Celery 任务：

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

5. 启动 Celery Worker： 启动 Celery worker 来处理任务：

   celery -A tasks worker --loglevel=info

6. 运行 FastAPI 应用： 启动 FastAPI 应用：

   uvicorn main:app --reload

本题详细解读

1. 为什么使用 Celery？

Celery 是一个分布式任务队列，用于处理异步任务和定时任务。在 FastAPI 中，Celery 可以帮助处理耗时操作，如发送邮件、处理大量数据等，从而避免阻塞主线程，提高应用的响应速度。

2. Celery 的配置

在 celery.py 文件中，我们配置了 Celery 实例，指定了消息代理（broker）和结果存储（backend）。这里使用了 Redis 作为消息代理和结果存储，但你也可以选择其他如 RabbitMQ 或数据库。

3. 定义任务

在 tasks.py 文件中，我们定义了一个简单的加法任务 add。通过 @app.task 装饰器，Celery 知道这是一个需要异步执行的任务。

4. 在 FastAPI 中调用任务

在 FastAPI 的路由中，我们通过 add.delay(x, y) 调用 Celery 任务。delay 方法将任务放入队列，并立即返回一个任务 ID。这个任务 ID 可以用于后续查询任务状态或结果。

5. 启动 Celery Worker

Celery worker 是实际执行任务的进程。通过 celery -A tasks worker --loglevel=info 命令启动 worker，它会从消息代理中获取任务并执行。

6. 运行 FastAPI 应用

最后，通过 uvicorn main:app --reload 启动 FastAPI 应用。此时，FastAPI 应用可以与 Celery 协同工作，处理异步任务。

通过以上步骤，你可以在 FastAPI 中成功集成 Celery，实现异步任务处理。