FastAPI与Celery联动：打造高效异步任务处理佳构

FastAPI与Celery联动：构建高效异步任务处理体系

一、Celery基础概念

Celery的架构由三部分构成：客户端用于发送任务、消息代理存储任务队列、工作者执行任务。常见的消息代理可选Redis（默认端口6379）或RabbitMQ（默认端口5672）。任务结果的存储建议使用Redis或关系型数据库，需在配置中明确指定backend参数。

二、FastAPI与Celery集成步骤

用户发起请求后，经FastAPI路由处理将任务入队，任务由消息队列传递给Celery工作者执行，执行完的结果存储，客户端可查询结果。

三、代码实现与解析

1. 安装依赖

通过以下命令安装所需依赖：

pip install fastapi==0.103.2 celery==5.3.4 redis==4.5.5 uvicorn==0.23.2 pydantic==2.5.2

2. 项目结构

项目结构如下：

project/
├── main.py
├── celery_app.py
└── tasks.py

3. 核心代码示例

celery_app.py

from celery import Celery

celery_app = Celery(
    'worker',
    broker='redis://localhost:6379/0',
    backend='redis://localhost:6379/1',
    include=['project.tasks']
)
celery_app.conf.update(task_track_started=True)

tasks.py

from .celery_app import celery_app


@celery_app.task
def process_data(data: dict) -> dict:
    """模拟耗时数据处理任务"""
    import time
    time.sleep(5)
    return {"result": f"Processed {data['input']}"}

main.py

from fastapi import FastAPI
from pydantic import BaseModel
from .tasks import process_data

app = FastAPI()


class RequestData(BaseModel):
    input: str
    priority: int = 1


@app.post("/submit-task")
async def submit_task(data: RequestData):
    """提交异步任务接口"""
    task = process_data.apply_async(
        kwargs={"data": data.dict()},
        priority=data.priority
    )
    return {"task_id": task.id}

四、任务状态查询实现

@app.get("/task-status/{task_id}")
async def get_task_status(task_id: str):
    """任务状态查询接口"""
    from celery.result import AsyncResult
    result = AsyncResult(task_id, app=celery_app)
    return {
        "status": result.status,
        "result": result.result if result.ready() else None
    }

五、课后Quiz

Q1：为何需为Celery配置不同Redis数据库作为broker和backend？
A1：为区分存储用途，避免任务队列与结果数据相互干扰。broker用0号库存放置理中的任务，backend用1号库保存任务状态和结果。

Q2：如何处理长时间运行任务的超时问题？
A2：在任务装饰器中设置soft_time_limit参数，例如@task(soft_time_limit=300)，同时配置worker的--maxtasksperchild参数限制最大任务数。

六、常见报错解决方案

错误现象：Worker收到任务但未执行

检查要点：
1. 确认Redis服务运行状态：执行redis-cli ping应返回PONG。
2. 检查任务模块是否在配置的include中正确指向。
3. 验证任务函数是否用@celery_app.task装饰器定义。

错误现象：任务结果无法获取

解决方案：
1. 检查backend配置是否正确。
2. 确认任务已执行完成（状态为SUCCESS）。
3. 检查Redis存储空间是否充足。

任务优先级配置

启动worker时指定队列：

celery -A project.celery_app worker -Q high_priority,default -l INFO

接口调用时指定优先级：

process_data.apply_async(
    kwargs={"data": data},
    queue='high_priority' if data.priority > 5 else 'default'
)

往期文章归档（示例）

• 如何用WebSocket构建毫秒级实时协作系统？ - cmdragon's Blog
• FastAPI中WebSocket消息处理的实用方法
• FastAPI安全防护的多层实践

免费在线工具（示例）

• CMDragon 在线工具 - 多功能AI工具箱与开发者工具集 | 免费实用工具
• 应用商店 - 搜罗千款提升效率与开发的AI工具与实用程序 | 免费工具
• AI文本生成图像工具 - 应用商店 | 免费工具