FastAPI 与异步服务开发实战

FastAPI 自带异步支持，但写好异步服务不只是把 def 改成 async def。从异步路由、后台任务到 WebSocket 与异步数据库驱动，本文讲清 FastAPI 异步服务开发的完整模式与常见坑。

一、为什么 FastAPI 天生异步

FastAPI 底层基于 Starlette 和 Uvicorn，天然运行在 asyncio 事件循环上。这意味着你写的路由函数如果是 async def，它就在事件循环里直接执行；如果是普通 def，FastAPI 会自动把它放到线程池里跑。理解这个区别是写好异步 FastAPI 的第一步。

from fastapi import FastAPI
import asyncio

app = FastAPI()

# 异步路由：直接在事件循环执行
@app.get("/async")
async def async_endpoint():
    await asyncio.sleep(1)  # 不阻塞事件循环
    return {"mode": "async"}

# 同步路由：FastAPI 自动放到线程池
@app.get("/sync")
def sync_endpoint():
    import time
    time.sleep(1)  # 阻塞线程但不阻塞事件循环
    return {"mode": "sync"}

二、异步数据库操作

用同步驱动（如 psycopg2）查数据库会阻塞事件循环。异步服务要用异步驱动（如 asyncpg、SQLAlchemy 2.0 async）。

from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from sqlalchemy import text

DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/db"
engine = create_async_engine(DATABASE_URL)
async_session = async_sessionmaker(engine, expire_on_commit=False)

app = FastAPI()

async def get_db():
    async with async_session() as session:
        yield session

@app.get("/users/{user_id}")
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(text("SELECT * FROM users WHERE id = :id"), {"id": user_id})
    user = result.first()
    return {"user": user}

三、后台任务

有些操作不需要等它完成再返回响应，比如发邮件、写日志。FastAPI 的 BackgroundTasks 可以在响应返回后异步执行。

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def send_email(email: str, message: str):
    # 模拟发邮件（实际用 smtplib 或 aiosmtplib）
    print(f"给 {email} 发送: {message}")

@app.post("/register")
async def register(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(send_email, email, "欢迎注册！")
    return {"status": "ok", "message": "注册成功，确认邮件已发送"}

客户端立即收到响应，发邮件在后台慢慢跑。注意 BackgroundTasks 是同进程内的轻量任务，如果任务很重或需要重试，应该用 Celery 或 RQ。

四、WebSocket 实时通信

FastAPI 原生支持 WebSocket，适合聊天、实时通知、数据流推送。

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()

class ConnectionManager:
    def __init__(self):
        self.active = []

    async def connect(self, ws: WebSocket):
        await ws.accept()
        self.active.append(ws)

    def disconnect(self, ws: WebSocket):
        self.active.remove(ws)

    async def broadcast(self, message: str):
        for ws in self.active:
            await ws.send_text(message)

manager = ConnectionManager()

@app.websocket("/ws/chat")
async def websocket_chat(ws: WebSocket):
    await manager.connect(ws)
    try:
        while True:
            data = await ws.receive_text()
            await manager.broadcast(f"收到: {data}")
    except WebSocketDisconnect:
        manager.disconnect(ws)

五、并发请求处理

在一个路由里需要同时调多个外部 API 时，用 asyncio.gather 并发执行，比串行快很多。

import httpx
import asyncio
from fastapi import FastAPI

app = FastAPI()

@app.get("/aggregate")
async def aggregate():
    async with httpx.AsyncClient() as client:
        # 并发请求三个 API
        results = await asyncio.gather(
            client.get("https://api.example.com/users"),
            client.get("https://api.example.com/products"),
            client.get("https://api.example.com/orders"),
        )
    return {
        "users": results[0].json(),
        "products": results[1].json(),
        "orders": results[2].json(),
    }

六、常见坑与最佳实践

坑一：async 路由里调阻塞代码。在 async def 里用 requests.get() 会阻塞整个事件循环。改用 httpx.AsyncClient。

坑二：忽略连接池。高并发下不要每次请求都新建数据库连接，用连接池（SQLAlchemy async engine 自带连接池）。

坑三：测试用同步客户端。测试异步路由要用 httpx.AsyncClient 或 FastAPI 的 TestClient（它内部自动处理异步）。

七、总结

FastAPI 的异步能力是它相比 Flask 的最大优势。正确使用方式：IO 密集操作用异步（HTTP 请求、数据库查询、WebSocket），CPU 密集操作用同步路由交给线程池。记住一个判断标准：你的代码里有没有等待外部资源？有就用 async，没有就用 def。