生产部署

从开发到生产要解决五件事：多进程运行、流式输出、实时通道、异步任务、流量治理。面试主线按这个顺序过一遍。

uvicorn 与多进程

# 开发
uvicorn main:app --reload

# 生产: 多 worker
uvicorn main:app --workers 4

# 经典组合: gunicorn 管理进程, uvicorn worker 跑 ASGI
gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000

方案	说明
单 worker	开发/低流量，单进程单事件循环
多 worker	利用多核，进程数 ≈ CPU 核数
gunicorn + uvicorn worker	生产经典：gunicorn 管进程生命周期，uvicorn 管 ASGI

要点：worker 间内存不共享（每进程独立事件循环）；有状态的东西（内存缓存、WebSocket 连接、定时任务）要外置（Redis、消息队列），否则多 worker 下行为错乱。反向代理（Nginx）在最外层。

流式响应

from fastapi.responses import StreamingResponse

@app.get("/stream")
async def stream():
    async def gen():
        for chunk in large_data():
            yield chunk
    return StreamingResponse(gen(), media_type="text/plain")

SSE（Server-Sent Events）：服务端单向推送，text/event-stream，聊天补全、进度通知
大文件下载：流式返回，不整块进内存
流式响应不适用：需要全局响应头/压缩的场景（内容长度未知）

WebSocket

@app.websocket("/ws")
async def ws(websocket: WebSocket):
    await websocket.accept()
    while True:
        msg = await websocket.receive_text()
        await websocket.send_text(f"echo: {msg}")

双向实时通道：聊天、协作编辑、实时通知
多 worker 下 WebSocket 连接粘在某个 worker 上，跨实例广播要 pub/sub（Redis）
生产规模：连接多时考虑独立网关或换 Go/Node 服务

后台任务

from fastapi import BackgroundTasks

def send_email(user_id: int):
    ...

@app.post("/users")
def create_user(user: UserCreate, tasks: BackgroundTasks):
    tasks.add_task(send_email, user.id)      # 响应返回后执行
    return {"ok": True}

方案	定位
BackgroundTasks	轻量：响应后跑一次，进程内执行（多 worker 下不保证执行节点）
APScheduler	定时任务
Celery / 消息队列	重任务：可靠投递、重试、任务队列

选型：邮件通知这种轻量活 BackgroundTasks 够；需要重试和可靠性的异步任务上 Celery/RQ。

限流

# slowapi: FastAPI 限流扩展
from slowapi import Limiter
limiter = Limiter(key_func=get_remote_address)

@app.get("/api")
@limiter.limit("100/minute")
def api(request: Request):
    ...

应用层限流（slowapi）：按 IP/用户限速，基于 Redis 存储跨实例共享计数
网关层限流（Nginx/API 网关）：全局流量治理，第一道防线
生产是两层都有：网关挡大流量，应用层做精细化控制

面试追问

多 worker 要注意什么？ 进程间不共享内存：缓存、WebSocket 连接、定时任务要外置（Redis/MQ）
SSE 和 WebSocket 的区别？ SSE 单向服务端推送（HTTP 长连接），WebSocket 双向。推送通知用 SSE，交互用 WS
BackgroundTasks 和 Celery？ 前者轻量进程内响应后执行，不保证可靠；后者分布式队列、重试、可靠投递
WebSocket 多实例怎么办？ 连接粘在单 worker，广播用 Redis pub/sub 跨实例分发
限流放哪层？ 网关层全局第一道（Nginx），应用层精细化（slowapi + Redis 计数）。两层叠加