FastAPI 生产部署实战：JWT 认证、Redis 限流与 Docker 零停机发布

FastAPI 接口在本地跑得好好的，一部署到线上，问题就开始冒头了。

接口被人反复刷怎么办？用户传个参数就能访问别人的数据怎么办？新版本发布时，正在处理的请求突然断了怎么办？

这些问题单看都不复杂，但真到了生产环境，光把接口写通远远不够。我自己做 Python 服务时，会优先把认证、限流和发布这三件事处理好。业务代码可以慢慢迭代，这几个地方没兜住，出问题往往不是改一行代码那么简单。

一、认证：别只靠一个 Token 字符串

先看一个常见问题。

有些项目为了省事，直接在请求头里判断：

if request.headers.get("token") == "admin123":
    return await handle_request()

这种写法连最基本的安全边界都没有。Token 写在代码里，泄漏后无法方便地单独吊销，也没有过期机制。

生产环境可以使用 OAuth2 Bearer Token 配合 JWT。下面用 FastAPI 和 PyJWT 实现一个简单的接口认证。

import os
import jwt
from datetime import datetime, timedelta, timezone
from fastapi import Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer

JWT_KEY = os.environ["JWT_SECRET"]
JWT_ALG = "HS256"

bearer = OAuth2PasswordBearer(tokenUrl="/auth/token")

def create_token(user_id: str) -> str:
    now = datetime.now(timezone.utc)
    payload = {
        "sub": user_id,
        "iat": now,
        "exp": now + timedelta(minutes=30),
    }
    return jwt.encode(payload, JWT_KEY, algorithm=JWT_ALG)

async def current_user(token: str = Depends(bearer)):
    try:
        payload = jwt.decode(
            token,
            JWT_KEY,
            algorithms=[JWT_ALG],
        )
        user_id = payload.get("sub")
        if not user_id:
            raise ValueError("missing subject")
        return user_id
    except (jwt.InvalidTokenError, ValueError):
        raise HTTPException(
            status_code=401,
            detail="认证失败",
            headers={"WWW-Authenticate": "Bearer"},
        )

接口里这样使用：

@app.get("/api/orders")
async def list_orders(user_id: str = Depends(current_user)):
    return await order_service.list_by_user(user_id)

这里有个细节：认证通过不代表授权通过。查询订单时，必须用当前用户身份约束数据库查询，不能只验证 Token，然后让客户端传什么 user_id 就查什么。

另外，JWT 的签名密钥要放在环境变量或密钥管理系统里，不要提交到 Git。真实项目还需要补齐登录密码校验、刷新令牌、主动吊销等机制。

二、限流：别等接口被刷了才想起来

认证解决的是“谁能访问”，限流解决的是“一个客户端能打多少请求”。

FastAPI 是异步框架，但异步不代表资源无限。一个接口如果每次都要查数据库、调用外部 API，即使单个请求耗时不长，大量并发请求照样会把连接池耗尽。

单进程服务可以先用 Redis 做限流。这里采用固定时间窗口，按用户维度统计请求数：

import os
from datetime import datetime, timezone

from fastapi import Request, HTTPException
from redis.asyncio import Redis

redis = Redis.from_url(
    os.environ["REDIS_URL"],
    decode_responses=True,
)

async def rate_limit(request: Request):
    user_id = getattr(request.state, "user_id", None)
    if not user_id:
        raise HTTPException(401, "请先登录")

    window = int(datetime.now(timezone.utc).timestamp()) // 60
    key = f"api:limit:{user_id}:{window}"

    count = await redis.incr(key)
    if count == 1:
        await redis.expire(key, 65)

    if count > 120:
        raise HTTPException(
            status_code=429,
            detail="请求过于频繁，请稍后再试",
            headers={"Retry-After": "60"},
        )

这段代码的思路很简单：每个用户每分钟最多允许 120 次请求。实际接入时，要确保认证中间件先把用户身份写入 request.state.user_id，再执行限流逻辑。

不过，别急着把它当成最终方案。固定窗口存在边界突刺问题，窗口切换时可能连续放过两批请求。对支付、短信发送等敏感接口，我更倾向于使用 Redis Lua 实现原子限流，或者采用令牌桶算法。

还有一个容易遗漏的地方：Redis 如果超时了怎么办？普通查询接口可以根据业务风险决定是否降级，但登录、短信发送等容易被滥用的接口，不应该因为限流组件故障就直接放开。限流规则也不能只按用户划分，还要考虑 IP、接口和全局并发量。

三、零停机发布：关键不在 FastAPI 本身

认证和限流解决了请求进来的问题，发布阶段还有另一类风险。

假设线上运行着旧版本，新版本已经部署完成。如果直接杀掉旧进程，再启动新进程，中间就可能出现连接失败、请求中断，甚至健康检查通过了，但数据库连接还没准备好的情况。

我的做法是让应用运行在多个独立进程或容器后面，由 Nginx、负载均衡器或容器编排平台负责流量切换。

FastAPI 应用先提供健康检查：

from fastapi import FastAPI, HTTPException
from redis.asyncio import Redis

app = FastAPI()

@app.get("/health/live")
async def live():
    return {"status": "ok"}

@app.get("/health/ready")
async def ready():
    try:
        await redis.ping()
        await order_service.check_database()
        return {"status": "ready"}
    except Exception:
        raise HTTPException(503, "服务尚未就绪")

这里要区分存活检查和就绪检查。存活检查用于判断进程是否还活着；就绪检查用于判断实例能否接收流量。数据库短暂故障时是否摘除实例，需要结合业务依赖和故障范围判断，不能简单地把所有依赖异常都当成进程死亡。

发布时采用滚动更新：先启动新实例，确认就绪，再把流量切过去；旧实例停止接收新请求后，等待已有请求完成，再退出。

例如使用 Docker Compose 配合 Nginx，可以让新旧容器短暂并存，再逐步切换上游。

若使用 Kubernetes，则可以通过 Deployment 的滚动更新策略和 readiness probe 实现类似流程。

需要注意，仅仅配置多个 Uvicorn Worker，并不等于零停机发布。Worker 数量解决的是单个实例的并发处理能力，发布期间的流量切换、优雅退出和实例替换，需要部署层共同配合。

数据库变更也要遵循兼容原则。新增字段时，先增加可空字段，再发布兼容新旧结构的代码，最后迁移数据、收紧约束。不要在新旧版本并存时直接删除旧字段，否则应用层没报错，数据库先把自己绊倒了。

四、上线前，至少把这几项检查一遍

我不会因为接口返回了 HTTP 200，就认定服务已经可以上线。

• 认证：缺少 Token、Token 过期、签名错误时，是否都能正确拒绝请求？
• 限流：Redis 超时、突发请求、多个实例同时处理时，是否有明确的降级策略？
• 发布：新实例启动失败时，旧实例是否仍然正常接流量？旧实例退出时，长请求有没有被粗暴中断？
• 监控：有没有记录请求耗时、HTTP 5xx、429、数据库连接池使用率，以及实例重启次数？
• 回滚：新版本出现问题，能否快速恢复旧版本？数据库变更是否兼容回滚？

这几项里，我尤其在意发布失败后的回滚。上线顺利不代表方案可靠，真正能看出发布设计有没有问题的，是新版本刚接流量就出错时，系统能不能稳住。

FastAPI 负责把接口写得清晰、高效，但认证、限流和零停机发布不是加几个装饰器就能全部解决的事情。应用代码要守住安全边界，Redis 要扛住限流计数，部署平台要保证流量切换，数据库变更还得兼容新旧版本。

这几层配合好了，才算真正把一个 FastAPI 服务送上生产环境。如果你也在处理类似的线上部署问题，欢迎到 云栈社区 一起交流。