FastAPI异步任务队列Celery完全指南

📂 所属阶段：第六阶段 — 2026 特色专题（AI 集成篇）
🔗 相关章节：流式响应 StreamingResponse · Redis 集成

异步任务队列：为什么我们需要它？

想象一下AI画头像、批量发周报、转码10GB视频这类场景——如果用同步Web接口处理：

  • 用户得盯着浏览器转几十秒甚至分钟圈
  • 服务器会被几个请求卡死
  • 超时后连重试机会都难抓

异步任务队列的核心价值就在于把「耗时/高风险操作」和「用户请求响应」彻底解耦：

sequenceDiagram
    participant User as 用户
    participant FastAPI as FastAPI(Producer)
    participant Broker as Broker(Redis/RabbitMQ)
    participant Worker as Celery Worker
    participant Backend as Backend(Redis)
    
    User->>FastAPI: 生成一张动漫头像
    FastAPI->>Broker: 提交任务到队列
    FastAPI->>User: 立即返回「任务ID+预估完成时间」
    Broker->>Worker: 分配空闲Worker
    Worker->>Backend: 存储进度/结果
    User->>FastAPI: 轮询/WebSocket查状态
    FastAPI->>Backend: 获取最新状态
    Backend->>FastAPI: 返回头像URL
    FastAPI->>User: 展示结果

快速上手：3步搭好基础架构

1. 安装核心依赖

# 使用Redis当Broker和Backend（最常用组合）
pip install fastapi uvicorn celery[redis] redis

2. 极简版配置与任务定义

# celery_app.py
from celery import Celery
import time

# 创建Celery实例
celery = Celery(
    'daoman_fastapi_celery',
    broker='redis://localhost:6379/0',  # Broker数据库0
    backend='redis://localhost:6379/1'   # Backend数据库1（分离隔离）
)

# 定义一个基础邮件任务
@celery.task(bind=True, name="tasks.send_email")
def send_email(self, to: str, subject: str, body: str):
    """
    模拟异步发邮件，带进度上报
    """
    try:
        self.update_state(state='PROGRESS', meta={'step': 'preparing'})
        time.sleep(1)
        
        self.update_state(state='PROGRESS', meta={'step': 'sending'})
        time.sleep(2)
        
        return {'status': 'success', 'to': to, 'time': time.time()}
    except Exception as exc:
        raise self.retry(exc=exc, countdown=60, max_retries=3)

3. FastAPI集成+启动流程

# main.py
from fastapi import FastAPI, HTTPException
from celery.result import AsyncResult
from celery_app import celery, send_email
from pydantic import BaseModel, EmailStr

app = FastAPI(title="道满FastAPI-Celery教程")

# 请求模型
class EmailReq(BaseModel):
    to: EmailStr
    subject: str
    body: str

# 提交任务接口
@app.post("/tasks/email")
async def create_email_task(req: EmailReq):
    task = send_email.delay(req.to, req.subject, req.body)
    return {
        "task_id": task.id,
        "status": "submitted",
        "message": "邮件已进入队列，预计3秒左右完成"
    }

# 查询任务状态接口
@app.get("/tasks/{task_id}")
async def get_task_status(task_id: str):
    task = AsyncResult(task_id, app=celery)
    return {
        "task_id": task_id,
        "state": task.state,
        "info": task.info if task.state in ['PROGRESS', 'FAILURE'] else None,
        "result": task.result if task.state == 'SUCCESS' else None
    }

启动顺序：

# 1. 启动Redis（本地调试用）
redis-server

# 2. 启动Celery Worker（终端2）
celery -A celery_app worker --loglevel=INFO --concurrency=4

# 3. 启动FastAPI（终端3）
uvicorn main:app --reload

核心进阶：企业级常用配置

1. 完整的Celery配置文件

# config/celery_config.py
from kombu import Queue
import os

class CeleryConfig:
    # Broker/Backend
    BROKER_URL = os.getenv('CELERY_BROKER', 'redis://localhost:6379/0')
    RESULT_BACKEND = os.getenv('CELERY_BACKEND', 'redis://localhost:6379/1')
    
    # 序列化（生产用msgpack，调试用json）
    TASK_SERIALIZER = 'json'
    ACCEPT_CONTENT = ['json']
    RESULT_SERIALIZER = 'json'
    
    # 时区（中国上海）
    TIMEZONE = 'Asia/Shanghai'
    ENABLE_UTC = False
    
    # 任务控制
    TASK_TRACK_STARTED = True  # 跟踪STARTED状态
    TASK_TIME_LIMIT = 300       # 硬超时5分钟（强制杀进程）
    TASK_SOFT_TIME_LIMIT = 240  # 软超时4分钟（抛异常可处理）
    
    # 重试策略
    TASK_RETRY_BACKOFF = True   # 指数退避（防雪崩）
    TASK_RETRY_BACKOFF_MAX = 700
    TASK_RETRY_JITTER = True    # 随机抖动（避免同时重试）
    
    # Worker优化
    WORKER_PREFETCH_MULTIPLIER = 1  # 防任务积压
    WORKER_MAX_TASKS_PER_CHILD = 1000  # 防内存泄漏
    WORKER_DISABLE_RATE_LIMITS = True  # 提升性能
    
    # 队列路由（分离不同任务）
    TASK_QUEUES = (
        Queue('default', routing_key='default'),
        Queue('ai_tasks', routing_key='ai.#'),
        Queue('email_tasks', routing_key='email.#'),
    )
    TASK_ROUTES = {
        'tasks.ai.*': {'queue': 'ai_tasks'},
        'tasks.email.*': {'queue': 'email_tasks'},
    }

2. 定时任务配置（Celery Beat）

# celery_beat.py
from celery.schedules import crontab
from celery_app import celery
import os

celery.conf.beat_schedule = {
    # 每天凌晨3点清理过期临时文件
    "cleanup-temp-files": {
        "task": "tasks.maintenance.cleanup_temp",
        "schedule": crontab(hour=3, minute=0),
        "options": {"queue": "maintenance"}
    },
    # 每15分钟检查队列状态
    "check-queue-health": {
        "task": "tasks.monitoring.check_queue",
        "schedule": crontab(minute="*/15"),
        "options": {"queue": "monitoring"}
    }
}

生产环境关键要点

1. 监控告警（Flower）

Flower是官方的Web监控工具，可以实时看任务、Worker状态：

# 安装并启动Flower
pip install flower
celery -A celery_app flower --port=5555 --basic_auth=admin:123456

2. Docker部署（单文件worker）

# Dockerfile.worker
FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .

# 创建非root用户
RUN useradd -m celery
USER celery

# 启动命令（指定队列）
CMD ["celery", "-A", "celery_app", "worker", "--loglevel=INFO", "--queues=default,ai_tasks"]

总结

Celery是FastAPI生态中最成熟的异步任务队列方案：

  • ✅ 解耦用户请求和耗时操作
  • ✅ 支持水平扩展Worker
  • ✅ 完善的重试、死信队列、定时任务
  • ✅ 丰富的监控工具（Flower）

常见问题FAQ

Q1: 任务提交后没反应？

A: 检查Redis是否启动、Worker是否连接正确的Broker、任务路由是否匹配队列。

Q2: 如何实现任务优先级？

A: 配置不同优先级的队列，高优先级队列启动更多Worker，或者使用Redis的RPUSHX和LPOP来手动控制（不推荐原生方式）。

🔗 相关教程