你写了个 FastAPI 接口，用户一请求，得等好几秒才返回，是不是急得想摔键盘？😤 别急着换电脑，八成是你把“耗时任务”堵在了接口里。

今天咱们就聊聊怎么用 Celery 这个老牌异步神器，给 FastAPI 提提速，顺便把我当年踩过的那些坑（比如循环导入、worker 启动失败）一次性说清楚。

🎯 核心脉络

💥 一、痛点：你的接口为什么这么慢？

有天你写了个 API，用户传个文件上来，你要做压缩、加水印、还要发邮件通知。一套下来可能 5 秒起步。这时候用户直接卡在“加载中”……

你可能会说：“加个 async def 不就行了？” 哎，单纯的 async/await 只能解决 IO 等待，治不了 CPU 密集型的活。这时候就需要一个真正的“后台打工仔”——Celery。

🧠 二、Celery 是个啥？用个比喻你就懂了

你把 FastAPI 想象成餐厅的前台点菜员，用户一来，点菜员记下菜单。

但做菜太慢，不能让客人站在前台等。所以 Celery 就是后厨的厨师团队：你丢个任务进去（比如“做一份宫保鸡丁”），厨师慢慢做，做完放传菜口。

前台可以马上给客人一个“你的菜在做啦”的号牌，过会儿凭号牌来取。

中间还得有个“订单板”——那就是 消息队列（Redis 或 RabbitMQ）。前台把任务写上去，厨师从上面拿。是不是瞬间清晰了？

🛠️ 三、实战：从零搭一套 FastAPI + Celery

好，咱们动手。这里千万别学我当初偷懒跳过环境准备，直接 pip install 就跑，后面一堆版本冲突的坑。

📦 1. 安装依赖

打开终端，一条龙装好：

uv add fastapi uvicorn celery redis

我用 Redis 做 broker（消息队列），你如果喜欢 RabbitMQ 也行，但 Redis 更轻量，本地调试方便。

🏗️ 2. 目录结构（血的教训）

project/
├── app.py # FastAPI 应用
├── tasks.py # Celery 任务定义
├── config.py # 配置（别写死在代码里）
└── pyproject.toml

重要的事说三遍：不要把 Celery 实例和 FastAPI app 写在同一个文件里！ 不然循环导入会教你做人。

📝 3. 写配置文件 config.py

REDIS_URL = "redis://localhost:6379/0"

🧵 4. 定义 Celery 任务 tasks.py

from celery import Celery
from config import REDIS_URL

celery_app = Celery(
    "myapp",
    broker=REDIS_URL,
    backend=REDIS_URL
)

@celery_app.task
def add(a, b):
    import time
    time.sleep(5) # 模拟耗时计算
    return a + b

这里注意：celery_app 是独立的，别引用 FastAPI 的任何东西，否则后面 worker 启动会一脸懵。

🚀 5. FastAPI 中调用任务 app.py

from fastapi import FastAPI
from tasks import add

app = FastAPI()

@app.get("/sum")
async def compute(x: int, y: int):
    # 异步提交任务，立刻返回任务ID
    task = add.delay(x, y)
    return {"task_id": task.id, "status": "processing"}

@app.get("/result/{task_id}")
async def get_result(task_id: str):
    from celery.result import AsyncResult
    from tasks import celery_app
    
    task = AsyncResult(task_id, app=celery_app)
    if task.ready():
        return {"status": "completed", "result": task.result}

    return {"status": "pending"}

你看，用户请求 /sum?x=3&y=5 会立刻得到一个 task_id，然后轮询 /result/{task_id} 就能拿到结果。再也不怕接口超时了！

⚠️ 四、你一定会踩的几个坑（我先帮你踩了）

🔮 五、进阶思考：生产环境还能怎么玩？

你以为这样就完了？Celery 还有定时任务（beat），比如每天凌晨三点清理数据库。另外，任务失败了怎么办？可以配置重试：

@celery_app.task(bind=True, max_retries=3)
def unstable_task(self):
    try:
        # 可能失败的操作
    except Exception as e:
        self.retry(exc=e, countdown=60)

还有监控工具 Flower，跑起来就能看任务队列、成功失败数量，像逛淘宝一样方便。命令：celery -A tasks flower，然后浏览器打开 localhost:5555。

好了，今天这篇算是把我用 FastAPI + Celery 从入门到“被坑”再到“平坑”的心里话都倒出来了。你写代码的时候如果遇到其他奇葩报错，欢迎在评论区甩过来，我帮你看看。觉得有用的话，点个关注或者收藏，免得下次想用翻不到。毕竟… 掉坑的记忆总会模糊，但收藏夹里的文章不会跑。😄