FastAPI + Celery 异步任务队列实战：从环境搭建到进度监控的完整指南

摘要：在使用 FastAPI 构建高性能 API 时，如何处理耗时任务而不阻塞主线程？本文通过一个完整的 Demo，从零搭建 FastAPI + Celery + Redis 异步任务队列，涵盖任务提交、状态追踪、进度更新及错误处理，助你掌握后端解耦的核心技术。

🎯 为什么你需要异步任务队列？

在构建现代 Web 应用时，我们常面临一个架构挑战：如何平衡接口的极速响应与后台的繁重计算？

如果将发送邮件、生成 PDF 报表或调用重型 AI 模型等操作放在 HTTP 请求的主线程中，FastAPI 的异步优势将荡然无存，用户不得不面对漫长的等待甚至超时。

解决方案是“解耦”：

• FastAPI 负责优雅地接收指令，立即返回“任务已受理”。
• Redis 作为消息中间件（Broker），充当缓冲区的神经中枢。
• Celery 作为后台 Worker，在幕后无声地处理繁重逻辑。

本文将带你通过一个开箱即用的 Demo，实现从任务调度到结果轮询的完整闭环。

---

📂 项目结构

我们将创建一个简洁明了的项目结构，便于理解和扩展：

fastapi_celery_demo/
├── app.py              # FastAPI 主应用（接口层）
├── celery_app.py       # Celery 配置实例（核心配置）
├── tasks.py            # 任务函数定义（业务逻辑）
├── docker-compose.yml  # Redis 服务配置（基础设施）
└── requirements.txt    # Python 依赖

---

🛠️ 1. 环境准备

依赖安装 (requirements.txt)

fastapi==0.104.1
uvicorn[standard]==0.24.0
celery==5.3.4
redis==5.0.1

基础设施 (docker-compose.yml)

我们使用 Docker 快速启动一个 Redis 实例，作为 Celery 的 Broker 和 Backend。

version: "3.8"

services:
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    restart: unless-stopped

---

⚙️ 2. 核心代码实现

Celery 配置 (celery_app.py)

这是异步系统的“心脏”。我们配置了 Redis 连接，并开启了任务追踪和 UTC 时间支持。

from celery import Celery

# 创建 Celery 实例
app = Celery(
    "fastapi_celery_demo",
    broker="redis://localhost:6379/0",  # 消息中间件
    backend="redis://localhost:6379/0", # 结果存储
)

# 优化配置
app.conf.update(
    task_serializer="json",
    accept_content=["json"],
    result_serializer="json",
    timezone="Asia/Shanghai",
    enable_utc=True,
    task_track_started=True,      # 追踪任务开始状态
    task_time_limit=30 * 60,      # 硬超时限制
    task_soft_time_limit=25 * 60, # 软超时限制
)

# 自动发现任务模块
app.autodiscover_tasks(["tasks"])

任务定义 (tasks.py)

这里定义了三种典型场景：简单计算、带进度的长任务、以及模拟失败的任务。

import time
from celery_app import app

@app.task(name="tasks.add", bind=True)
def add(self, x: int, y: int) -> dict:
    """场景1：简单的加法任务"""
    self.update_state(state="PROCESSING", meta={"progress": "开始计算..."})
    time.sleep(2)  # 模拟耗时
    return {"task_id": self.request.id, "x": x, "y": y, "result": x + y}

@app.task(name="tasks.sleep", bind=True)
def sleep_task(self, seconds: int) -> dict:
    """场景2：模拟长任务，支持实时进度更新"""
    for i in range(seconds):
        time.sleep(1)
        self.update_state(
            state="PROGRESS",
            meta={
                "current": i + 1,
                "total": seconds,
                "progress": f"{(i + 1) / seconds * 100:.0f}%"
            }
        )
    return {"status": "done", "slept": seconds}

@app.task(name="tasks.fail", bind=True)
def fail_task(self) -> dict:
    """场景3：模拟任务失败"""
    time.sleep(1)
    raise ValueError("This task failed intentionally!")

FastAPI 接口 (app.py)

这是系统的“门面”，负责接收请求并返回任务 ID，同时提供查询接口。

from fastapi import FastAPI, HTTPException
from celery.result import AsyncResult
from pydantic import BaseModel
from celery_app import app
from tasks import add, sleep_task, fail_task

fastapi_app = FastAPI(title="FastAPI + Celery Demo")

class AddTaskRequest(BaseModel):
    x: int
    y: int

class SleepTaskRequest(BaseModel):
    seconds: int

@fastapi_app.post("/tasks/add")
async def create_add_task(request: AddTaskRequest):
    """提交加法任务"""
    task = add.delay(request.x, request.y)
    return {"task_id": task.id, "status": "submitted"}

@fastapi_app.post("/tasks/sleep")
async def create_sleep_task(request: SleepTaskRequest):
    """提交长时任务"""
    if request.seconds <= 0:
        raise HTTPException(status_code=400, detail="seconds must be > 0")
    task = sleep_task.delay(request.seconds)
    return {"task_id": task.id, "status": "submitted"}

@fastapi_app.get("/tasks/{task_id}")
async def get_task_status(task_id: str):
    """查询任务实时状态（含进度）"""
    task_result = AsyncResult(task_id, app=app)
    response = {"task_id": task_id, "status": task_result.state, "ready": task_result.ready()}
    
    if task_result.state == "PROGRESS":
        response["progress"] = task_result.info
    elif task_result.ready():
        if task_result.successful():
            response["result"] = task_result.result
        else:
            response["error"] = str(task_result.info)
    return response

---

🚀 3. 启动与测试

第一步：启动 Redis

docker-compose up -d

第二步：启动 Celery Worker

在新终端中运行：

celery -A celery_app worker --loglevel=info

看到 celery@your-host ready 即表示连接成功。

第三步：启动 FastAPI

再开一个新终端：

uvicorn app:fastapi_app --reload --host 0.0.0.0 --port 8000

第四步：交互式测试

访问 Swagger UI: http://localhost:8000/docs

1. POST /tasks/sleep: 输入 {"seconds": 5}，获取 task_id。
2. GET /tasks/{task_id}: 多次刷新，观察 status 从 PENDING -> PROGRESS (带百分比) -> SUCCESS 的变化。

---

💡 进阶技巧

• 监控面板：安装 flower (pip install flower) 并运行 celery -A celery_app flower，可以通过 Web 界面直观查看任务队列堆积情况。
• 并发控制：启动 Worker 时增加 --concurrency=4 参数，根据 CPU 核心数调整并行处理能力。
• 生产环境建议：在生产环境中，建议使用 Supervisor 或 systemd 管理 Celery Worker 进程，确保其崩溃后能自动重启。

---

🏁 结语

通过 FastAPI 与 Celery 的结合，我们成功将“即时响应”与“后台处理”分离。这种架构不仅提升了用户体验，也为系统未来的横向扩展打下了坚实基础。希望这个 Demo 能成为你构建高可用后端服务的有力起点。

喜欢这篇教程吗？欢迎在评论区分享你的异步任务实战经验！