FastAPI自定义后台任务队列教程

引言

FastAPI是一个现代、快速（高性能）的Web框架，基于Python类型提示，广泛用于构建API。在许多Web应用中，我们需要处理耗时操作，如发送电子邮件、数据处理或外部API调用。为了不阻塞HTTP请求响应，后台任务队列是解决此问题的关键。本教程将引导您如何在FastAPI中创建自定义后台任务队列，从基础概念到实际实现，适合新手学习。

什么是后台任务队列？

后台任务队列是一种异步处理机制，允许您将任务提交到队列中，由专门的“工作者”进程在后台处理。这样，主请求线程可以立即返回响应，而任务在后台异步执行，提升了用户体验和应用性能。

常见用例包括：

• 发送批量邮件
• 图像处理
• 数据导入/导出
• 定时任务调度

FastAPI内置后台任务

FastAPI提供了BackgroundTasks类，用于简单的后台任务处理。它通过依赖注入在请求处理完后执行任务，但仅限于当前请求的生命周期内，不适合持久化或复杂的队列管理。

示例代码：

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def send_email(email: str, message: str):
    # 模拟发送邮件
    print(f"Sending email to {email}: {message}")

@app.post("/send-email/")
async def create_email(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(send_email, email, "Welcome message")
    return {"message": "Email will be sent in background"}

这种方式简单易用，但不支持任务持久化、重试或分布式处理，因此对于复杂需求，我们需要自定义后台任务队列。

为什么需要自定义后台任务队列？

• 持久化：任务可以在应用重启后继续处理。
• 重试机制：处理任务失败时自动重试。
• 优先级和队列管理：支持多个队列和任务优先级。
• 分布式处理：在多台机器上分发任务，提高处理能力。
• 监控和日志：便于跟踪任务状态和性能。

实现自定义后台任务队列

在FastAPI中，常用第三方库如Celery或RQ来创建自定义后台任务队列。这里以Celery为例，因为它功能强大且社区支持广泛。RQ作为轻量级替代，步骤类似但更简单。

使用Celery

Celery是一个分布式任务队列，支持多种消息代理（如RabbitMQ、Redis）。我们将使用Redis作为消息代理，因为它易于安装和使用。

步骤1：安装依赖

确保已安装Python和pip，然后运行：

pip install fastapi celery redis uvicorn

这安装了FastAPI、Celery、Redis客户端和ASGI服务器。

步骤2：配置Redis

首先，安装并启动Redis服务器。可以从官网下载或使用Docker。例如，在本地运行：

redis-server

默认端口是6379。

步骤3：创建Celery应用

创建一个Python文件，如celery_app.py，配置Celery：

from celery import Celery

# 初始化Celery应用
celery_app = Celery(
    'tasks',  # 应用名称
    broker='redis://localhost:6379/0',  # 消息代理URL，使用Redis
    backend='redis://localhost:6379/0'  # 结果存储后端
)

# 可选：配置任务序列化和时区
celery_app.conf.update(
    task_serializer='json',
    accept_content=['json'],
    result_serializer='json',
    timezone='UTC',
    enable_utc=True,
)

步骤4：定义任务

在同一文件或另一个文件中定义后台任务函数，用@celery_app.task装饰器标记：

@celery_app.task
def process_data(data: str):
    # 模拟耗时操作
    import time
    time.sleep(2)
    print(f"Processing data: {data}")
    return {"status": "completed", "data": data}

步骤5：在FastAPI中集成

在FastAPI主应用中，导入Celery任务并使用它。创建main.py：

from fastapi import FastAPI
from celery_app import process_data

app = FastAPI()

@app.post("/process/")
async def start_processing(data: str):
    # 触发后台任务，使用.delay()方法异步执行
    task = process_data.delay(data)
    return {
        "message": "Processing started in background",
        "task_id": task.id  # 返回任务ID以便跟踪
    }

@app.get("/status/{task_id}")
async def get_status(task_id: str):
    # 检查任务状态
    from celery.result import AsyncResult
    result = AsyncResult(task_id, app=process_data.app)  # 注意：这里需要传递Celery应用实例
    return {"task_id": task_id, "status": result.status, "result": result.result}

步骤6：启动应用

1. 启动Redis服务器（如果还没运行）。
2. 启动Celery工作者处理任务：

celery -A celery_app worker --loglevel=info

3. 启动FastAPI应用：

uvicorn main:app --reload

现在，访问API端点（如/process/）提交任务，它会异步处理。

使用RQ（可选）

RQ（Redis Queue）是另一个轻量级任务队列库，适合简单场景。安装：

pip install rq

配置类似，但更简单。创建rq_worker.py定义任务，并在FastAPI中通过Redis队列提交任务。

示例项目

为了加深理解，可以创建一个完整的FastAPI项目，集成Celery处理电子邮件发送任务。步骤如下：

1. 项目结构：创建app/main.py、app/tasks.py等文件。
2. 在tasks.py中定义send_email任务。
3. 在FastAPI路由中调用任务。
4. 使用环境变量配置Redis连接。
5. 添加错误处理和日志记录。

这有助于实践所学知识。

总结和最佳实践

• 选择合适的库：根据需求选择Celery（复杂、分布式）或RQ（简单、轻量）。
• 监控和日志：使用工具如Flower（Celery监控）或RQ Dashboard来监控任务状态。
• 错误处理：在任务中添加重试逻辑和异常捕获。
• 安全性：确保消息代理（如Redis）有适当的访问控制。
• 测试：编写单元测试模拟后台任务执行。

通过本教程，您应该能够掌握在FastAPI中实现自定义后台任务队列的基本方法。从内置BackgroundTasks到第三方库集成，逐步提升应用能力。继续探索FastAPI文档和社区资源，以应对更高级场景。