FastAPI 性能优化怎么做？缓存和序列化有哪些坑？

简化版

FastAPI 号称「高性能」，但一个真实项目的瓶颈几乎从来不在框架本身——按出现频率排：① 数据库 N+1 和慢查询；② 在 async def 里写了阻塞代码（把整个事件循环卡住，比用 def 还慢几十倍）；③ 序列化开销（response_model 会「再校验一次 + jsonable_encoder + json.dumps」，大列表接口上可能占一半时间）；④ 同步调用外部服务且没设超时；⑤ worker/连接池配置不合理。所以优化顺序是「先测量 → 消灭 N+1 → 排查阻塞 → 减少数据量 → 优化序列化 → 加缓存 → 调 worker」，缓存排在很后面。FastAPI 特有的两个性能点：① 序列化——换 ORJSONResponse 能快 3~5 倍，大列表接口直接返回 JSONResponse 可以完全跳过 response_model 的处理链；② 依赖开销——依赖树在启动时构建（运行时很轻），但同一请求内相同依赖会被缓存只执行一次，而在依赖里做重活（每个请求都跑一遍的权限查询、配置加载）是隐蔽的性能杀手。缓存分四层：请求内用 request.state、进程内用 lru_cache（只适合不变的数据，多 worker 下每进程一份）、跨进程用 Redis、HTTP 层用 ETag/Cache-Control。用缓存有三个必须注意的：异步应用要用 redis.asyncio 而不是同步客户端（否则阻塞事件循环）、缓存 key 必须包含所有影响响应的因素（尤其是用户身份，否则会数据串号）、难的是失效不是写入。核心记忆：瓶颈不在框架；先测量再优化，缓存排最后；ORJSONResponse 和跳过 response_model；异步应用要用异步 Redis 客户端。

详细版

优化手段按性价比排序：

手段	典型收益	成本
消灭 N+1（预加载）	10~100 倍	低
修掉 async def 里的阻塞	10~50 倍	低
加索引	10~1000 倍	低
分页/减少字段	数倍	低
ORJSONResponse	序列化快 3~5 倍	极低（一行）
跳过 response_model	大列表明显	中（失去校验）
Redis 缓存	数倍~数十倍	中（失效复杂）
调 worker/连接池	线性	低

# ① ★★序列化优化：一行换 orjson★★
from fastapi.responses import ORJSONResponse
app = FastAPI(default_response_class=ORJSONResponse)    # ★全局★
@app.get("/items", response_class=ORJSONResponse)       # ★或单个路由★

# ② ★★大列表：跳过 response_model 的处理链★★
# ✗ 慢：每条都要 再校验 + jsonable_encoder + dumps
@app.get("/items", response_model=list[ItemOut])
async def list_items(db: DbDep):
    return await db.scalars(select(Item))               # ★1000 条 = 1000 次校验★

# ✓ 快：直接返回 Response（★FastAPI 检测到是 Response 就跳过序列化★）
@app.get("/items")
async def list_items_fast(db: DbDep):
    rows = (await db.execute(
        select(Item.id, Item.name, Item.price))).mappings().all()  # ★★不构造 ORM 对象★★
    return ORJSONResponse(content=[dict(r) for r in rows])
# ★代价：失去 response_model 的字段过滤和文档 → 要自己保证不泄露敏感字段★

# ③ ★依赖里的隐蔽开销★
# ✗ 每个请求都查一次数据库
async def get_settings_from_db(db: DbDep):
    return await db.scalar(select(Setting))             # ★★每请求一次 SQL★★
# ✓ 用 lru_cache（不变的数据）或 Redis（会变的）
@lru_cache(maxsize=1)
def get_settings() -> Settings: return Settings()

# ④ ★★异步 Redis（不能用同步客户端）★★
import redis.asyncio as aioredis                        # ★★注意是 asyncio 版★★
redis_pool = aioredis.from_url(settings.REDIS_URL,
                               max_connections=20,
                               decode_responses=True)
async def get_redis() -> aioredis.Redis:
    return redis_pool
RedisDep = Annotated[aioredis.Redis, Depends(get_redis)]

# ⑤ ★缓存装饰器（自己实现或用 fastapi-cache2）★
def cached(ttl: int = 60, key_builder=None):
    def deco(fn):
        @wraps(fn)
        async def wrapper(*args, **kwargs):
            key = key_builder(*args, **kwargs) if key_builder else \
                  f"{fn.__module__}.{fn.__name__}:{hash_args(args, kwargs)}"
            cached_val = await redis_pool.get(key)
            if cached_val is not None:
                return orjson.loads(cached_val)
            result = await fn(*args, **kwargs)
            await redis_pool.set(key, orjson.dumps(result), ex=ttl)
            return result
        return wrapper
    return deco

@cached(ttl=300, key_builder=lambda user_id, **_: f"stats:{user_id}")
async def get_user_stats(user_id: int): ...             # ★★key 含用户 ID★★

# ⑥ ★HTTP 缓存（挡在应用之外，收益最大）★
@app.get("/config")
async def config(request: Request):
    data = await get_config()
    etag = hashlib.md5(orjson.dumps(data)).hexdigest()
    if request.headers.get("if-none-match") == etag:
        return Response(status_code=304)                # ★★不传 body★★
    return ORJSONResponse(data, headers={
        "ETag": etag, "Cache-Control": "private, max-age=300"})

# ⑦ ★uvicorn/gunicorn 配置★
# gunicorn -w 4 -k uvicorn.workers.UvicornWorker \
#   --timeout 60 --max-requests 10000 --max-requests-jitter 1000 app:app
# ★或 uvicorn --workers 4 --loop uvloop --http httptools★

⚠️ 三个必须记住的点：① 优化的第一步永远是测量，而且要先排除「阻塞」这个最容易被忽略的原因。FastAPI 项目最典型的性能事故不是「查询慢」而是「在 async def 里写了同步阻塞代码」——requests.get()、同步的数据库 session、bcrypt.checkpw()、大图片处理，任何一个都会把整个事件循环卡住，让所有并发请求排队串行。典型症状是「CPU 占用很低但 QPS 上不去、P99 随并发线性恶化」——遇到这个现象先查阻塞，别急着加缓存。② response_model 的开销比想象中大。它不只是「按模型筛选字段」——FastAPI 会用这个模型重新校验一遍你返回的数据，然后 jsonable_encoder 递归转换，最后 json.dumps。返回 1000 条记录时这三步都要跑 1000 次。两个优化：换 ORJSONResponse（Rust 实现，快 3~5 倍，原生支持 datetime/UUID）、大列表直接返回 Response 实例（FastAPI 检测到返回值已经是 Response 就完全跳过序列化链，代价是失去字段过滤和自动文档，要自己保证不泄露敏感字段）。③ 异步应用里必须用异步的 Redis/HTTP 客户端。import redis 是同步客户端，在 async def 里调用 r.get() 会阻塞事件循环——加缓存本来是为了变快，结果因为用了同步客户端反而更慢。正确的是 import redis.asyncio as aioredis；HTTP 调用同理用 httpx.AsyncClient 而不是 requests。

完整版教学

一、先测量：瓶颈在哪

★ ★一个典型请求的时间分布★：
  ┌────────────────────────────────────────────────────┐
  │ ASGI + 路由匹配             │ ★< 0.1ms★              │
  │ 依赖解析                    │ ★0.1~1ms（简单树）★     │
  │ Pydantic 请求校验           │ ★0.1~2ms★              │
  │ ★数据库查询★                │ ★★5~2000ms★★           │
  │ ★外部 API 调用★             │ ★★50~5000ms★★          │
  │ ★序列化（1000 条）★          │ ★★10~100ms★★           │
  └────────────────────────────────────────────────────┘
  ★ ★框架本身的开销可以忽略★

★ ★★第一件事：排除阻塞★★
  ★症状：★CPU 空闲 + QPS 上不去 + P99 随并发线性恶化★
  ★ 检测手段：
    ① ★asyncio debug 模式★（最直接）
       import asyncio
       loop = asyncio.get_event_loop()
       loop.set_debug(True)
       loop.slow_callback_duration = 0.1
       # → ★"Executing <Task...> took 0.523 seconds" 警告★
       # uvicorn: PYTHONASYNCIODEBUG=1
    ② ★blockbuster 库★（自动检测阻塞调用并抛异常）
    ③ ★py-spy dump --pid X★（看线程栈卡在哪）
    ④ ★压测对比：并发 1 vs 并发 50 的 P99★
       ★如果并发 50 时延迟正好是 50 倍 → 完全串行 → 有阻塞★

★ ★第二件事：数据库★
  # 开 SQL 日志
  create_async_engine(url, ★echo=True★)
  # ★统计每个请求的 SQL 条数★
  from sqlalchemy import event
  @event.listens_for(engine.sync_engine, "before_cursor_execute")
  def count(conn, cursor, stmt, params, context, many):
      ★ctx_sql_count.set(ctx_sql_count.get(0) + 1)★     # ★contextvar★
  # 中间件里超阈值告警
  if ctx_sql_count.get(0) > 20:
      logger.warning("N+1? path=%s n=%d", path, ctx_sql_count.get())

★ ★工具★：
  ┌────────────────────┬──────────────────────────────┐
  │ ★py-spy★            │ ★不侵入的采样 profiler★       │
  │                     │ py-spy top --pid X            │
  │                     │ py-spy record -o out.svg      │
  │ ★cProfile★          │ 精确但有开销，找 CPU 热点      │
  │ ★pyinstrument★      │ ★调用树可视化，适合单请求分析★ │
  │ ★OpenTelemetry★     │ ★生产环境分布式追踪★          │
  │ ★locust / wrk / k6★ │ 压测                          │
  └────────────────────┴──────────────────────────────┘

★ ★最小可观测中间件★：
  class MetricsMiddleware:
      def __init__(self, app): self.app = app
      async def __call__(self, scope, receive, send):
          if scope["type"] != "http": return await self.app(scope, receive, send)
          t0 = time.perf_counter(); status = 500
          async def send_wrapper(msg):
              nonlocal status
              if msg["type"] == "http.response.start":
                  status = msg["status"]
                  dur = (time.perf_counter() - t0) * 1000
                  msg["headers"].append((b"x-process-time", f"{dur:.1f}".encode()))
              await send(msg)
          try:
              await self.app(scope, receive, send_wrapper)
          finally:
              metrics.histogram("http.duration",
                  (time.perf_counter()-t0)*1000,
                  tags={"path": scope["path"], "status": status})

★ ★压测的正确做法★：
  wrk -t4 -c100 -d30s --latency http://localhost:8000/api/items
  ★ 关注：★QPS、P50/P95/P99、错误率★
  ★ ✗ 别在开发机上压（数据量、配置都不一样）
  ★ ✓ ★对比不同并发下的 P99★（发现阻塞和池打满）

一个典型请求的时间分布里，框架本身的开销可以忽略（ASGI 加路由不到 0.1ms、依赖解析和校验各 1ms 上下），大头在数据库、外部调用和序列化。优化的第一件事是排除阻塞——症状是「CPU 空闲 + QPS 上不去 + P99 随并发线性恶化」；检测手段有四个：asyncio debug 模式（slow_callback_duration = 0.1 会打印「took 0.523 seconds」警告，最直接）、blockbuster 库、py-spy dump 看线程栈、以及压测对比并发 1 和并发 50 的 P99（如果正好差 50 倍就是完全串行）。第二件事是统计每个请求的 SQL 条数（用 before_cursor_execute 事件配 contextvar，超阈值告警）。压测要注意别在开发机上压，重点看不同并发下的 P99 变化。

二、序列化优化

★ ★response_model 的完整开销★：
  返回 obj
    ↓ ① ★用 response_model 重新校验★（★不只是筛字段！★）
    ↓ ② ★jsonable_encoder 递归转换★（datetime/Decimal/UUID/Enum/BaseModel）
    ↓ ③ ★json.dumps★
  ★ ★1000 条记录 = 这三步各跑 1000 次★

★ ★★优化一：ORJSONResponse（一行，最划算）★★：
  from fastapi.responses import ORJSONResponse
  app = FastAPI(default_response_class=ORJSONResponse)
  ★ orjson 的优势：
    - ★Rust 实现，序列化快 3~5 倍★
    - ★原生支持 datetime / UUID / dataclass / numpy★
    - ★输出更紧凑★（默认不加空格）
  ★ 注意：★orjson 输出 bytes 而不是 str★（FastAPI 已处理）
  ★ 也可以用 ★UJSONResponse★（略快但兼容性差些）

★ ★★优化二：跳过 response_model（大列表）★★：
  # FastAPI 的逻辑：
  if isinstance(raw_response, Response):
      return raw_response                     # ★★直接用，跳过所有序列化★★
  ★ 所以：
    return ORJSONResponse(content=data)       # ★零 response_model 开销★
  ★ ✗ 代价：
    - ★失去自动的字段过滤★（★可能泄露敏感字段★）
    - ★OpenAPI 文档里没有响应结构★（可以用 responses= 手动补）
    - ★失去了"契约校验"★
  ✓ 折中：★只在确实是瓶颈的大列表接口上用★，并且自己控制字段：
    rows = await db.execute(select(Item.id, Item.name))   # ★★只查需要的列★★
    return ORJSONResponse([dict(r) for r in rows.mappings()])

★ ★优化三：只查需要的列（★更根本★）★：
  # ✗ 加载完整 ORM 对象
  items = (await db.scalars(select(Item))).all()
  # ✓ ★只查列，不构造 ORM 对象★
  rows = (await db.execute(select(Item.id, Item.name, Item.price))).mappings().all()
  ★ 收益：★少查数据、少传输、少构造对象、少序列化★
  ★ 尤其是表里有大字段（正文、JSON）时

★ ★优化四：response_model_exclude_unset★：
  @app.get("/x", response_model=Model, ★response_model_exclude_unset=True★)
  ★ 只序列化"实际设置过的字段" → 响应更小
  ★ 相关：exclude_none / exclude_defaults / include / exclude

★ ★算例：返回 5000 条记录★：
  ┌──────────────────────────────────────┬──────────┐
  │ response_model + 默认 JSONResponse    │ ★~450ms★ │
  │ response_model + ORJSONResponse       │ ~280ms   │
  │ ★直接返回 ORJSONResponse（跳过 model）★│ ★~60ms★  │
  │ ★+ 只查需要的列★                       │ ★~35ms★  │
  │ ★+ 分页（返回 50 条）★                 │ ★~2ms★   │
  └──────────────────────────────────────┴──────────┘
  ★ ★结论：分页才是根本，序列化优化是次要的★

★ ★压缩★：
  app.add_middleware(GZipMiddleware, minimum_size=1000)
  ★ ✓ JSON 压缩率很高（★常能压到 10~20%★）
  ★ ✗ ★占用 CPU★、★破坏流式响应★
  ✓ ★交给 Nginx 做★（可以用 brotli，且不占应用 CPU）

response_model 的开销是「再校验 + jsonable_encoder + json.dumps」三步，1000 条记录就各跑 1000 次。三个优化按性价比：① ORJSONResponse 一行搞定，快 3~5 倍（Rust 实现，原生支持 datetime/UUID）；② 直接返回 Response 实例跳过整个序列化链（代价是失去字段过滤和文档，可能泄露敏感字段，只在确实是瓶颈时用）；③ 只查需要的列（更根本——少查、少传、少构造、少序列化）。那个 5000 条记录的算例说明了关键：从 450ms 优化到 35ms 靠的是「跳过 model + 只查列」，但分页直接降到 2ms——所以分页才是根本，序列化优化是次要的。压缩建议交给 Nginx（不占应用 CPU，还能用 brotli，而且应用层的 GZipMiddleware 会破坏流式）。

三、缓存的层次

★ ★四层缓存（从近到远）★：
  ┌────────────────────────────────────────────────────────┐
  │ ① ★请求内（request.state / contextvar）★  ★零成本★       │
  │ ② ★进程内（lru_cache）★    ★极快，但多 worker 不共享★     │
  │ ③ ★跨进程（Redis）★        ★大多数场景★                  │
  │ ④ ★HTTP（ETag/CDN）★       ★★请求根本不到应用，收益最大★★  │
  └────────────────────────────────────────────────────────┘

★ ★① 请求内缓存★：
  # FastAPI 的依赖缓存本身就是这个（use_cache=True 默认）
  async def get_current_user(...): ...       # ★一个请求内只执行一次★
  # 手动：
  async def get_config(request: Request):
      if not hasattr(request.state, "config"):
          request.state.config = await load_config()
      return request.state.config

★ ★② 进程内缓存（lru_cache）★：
  @lru_cache(maxsize=1)
  def get_settings() -> Settings: return Settings()
  @lru_cache(maxsize=512)
  def get_country_name(code: str) -> str: ...
  ★ ✓ 适合：★配置、字典表、编译好的正则、不变的映射★
  ★ ✗ 不适合：
    - ★会变的业务数据★（多 worker 下各进程不一致）
    - ★async 函数★（lru_cache 缓存的是 coroutine 对象，★只能 await 一次★！）
      ✗ @lru_cache
        async def f(): ...          # ★★第二次 await 会报错★★
      ✓ 用 ★async-lru★ 或自己实现
  ★ ★带 TTL 的技巧★：
    @lru_cache(maxsize=32)
    def _get(key, _bucket): ...
    def get(key): return _get(key, int(time.time()) // 60)   # ★每分钟换 key★

★ ★③ Redis（异步客户端！）★：
  import redis.asyncio as aioredis            # ★★不是 import redis★★
  pool = aioredis.from_url(url, max_connections=20)
  # ✗ import redis; r = redis.Redis()   → ★r.get() 阻塞事件循环★

  ★ 现成的库：★fastapi-cache2★
    from fastapi_cache import FastAPICache
    from fastapi_cache.backends.redis import RedisBackend
    from fastapi_cache.decorator import cache
    FastAPICache.init(RedisBackend(redis), prefix="myapp")
    @app.get("/items")
    ★@cache(expire=60)★
    async def items(): ...
    ★ ✗ 注意：★默认 key 包含完整 URL（含查询串），但★不含用户★★
      → ★★用户相关的数据必须自定义 key_builder★★

★ ★★④ HTTP 缓存（最被低估）★★：
  # ETag（内容没变就返回 304，不传 body）
  etag = f'W/"{hashlib.md5(orjson.dumps(data)).hexdigest()}"'
  if request.headers.get("if-none-match") == etag:
      return Response(status_code=304)
  return ORJSONResponse(data, headers={"ETag": etag,
                                       "Cache-Control": "private, max-age=60"})
  # 公开内容让 CDN 缓存
  headers={"Cache-Control": "public, max-age=300", "Vary": "Accept-Encoding"}
  ★ ★★危险：带用户数据的响应必须 private 或 no-store★★
    → ★否则 CDN 可能把 A 用户的数据发给 B★（★重大事故★）

★ ★缓存 key 的设计（★最容易出事的地方★）★：
  ★ ★key 必须包含所有影响响应内容的因素★：
    - ★用户 ID / 租户 ID★  ← ★★漏了就是数据串号★★
    - 查询参数（分页、筛选、排序）
    - 语言 / 时区
    - ★版本号（便于批量失效）★
  key = f"v{VERSION}:user:{uid}:items:{page}:{size}:{status}"

★ ★失效策略★：
  ① ★TTL 兜底★（最简单，有不一致窗口）
  ② ★写时主动删★（及时，但要想清楚删哪些 key）
  ③ ★版本号/代际★（★批量失效的优雅方案★）
     ver = await redis.get("items:ver") or 1
     key = f"items:v{ver}:{page}"
     # 写入时：await redis.incr("items:ver")  ← ★★旧 key 自然失效★★
  ★ ★推荐：写时主动删 + 较短 TTL 兜底★

四层缓存的原则是「能在更外层挡住的就别放内层」。lru_cache 有个 FastAPI 特有的坑：不能直接装饰 async 函数——它缓存的是 coroutine 对象，第二次 await 会报错，要用 async-lru。Redis 必须用 redis.asyncio（同步客户端会阻塞事件循环，加缓存反而更慢）。fastapi-cache2 的默认 key 包含完整 URL 但不含用户——用户相关的数据必须自定义 key_builder。HTTP 缓存最被低估（请求根本到不了应用），但带用户数据的响应必须标 private 或 no-store，否则 CDN 可能把 A 用户的数据发给 B。缓存 key 的设计是最容易出事的地方——必须包含所有影响响应内容的因素，漏了用户 ID 就是数据串号；批量失效推荐用版本号/代际（incr 一下旧 key 自然失效）。

四、并发与资源配置

★ ★worker 与线程池★：
  # ★进程数★
  gunicorn -w 4 -k uvicorn.workers.UvicornWorker app:app
  # 或 uvicorn --workers 4
  ★ ★workers ≈ CPU 核数★（IO 密集可以略多）
  ★ ★每个 worker 是独立进程：独立的事件循环 + 独立的 40 线程池★

  # ★线程池（def 路由和文件 IO 用的）★
  @asynccontextmanager
  async def lifespan(app):
      anyio.to_thread.current_default_thread_limiter().total_tokens = 100
      yield

★ ★★连接数的算术（最容易爆）★★：
  ┌──────────────────────────────────────────────────┐
  │ ★数据库连接★ = workers × (pool_size + max_overflow)│
  │   4 × (10 + 20) = ★120★                           │
  │   PostgreSQL 默认 max_connections = ★100★ → ★爆★  │
  │ ★Redis 连接★ = workers × max_connections           │
  │ ★HTTP 连接★  = workers × httpx 的 limits           │
  └──────────────────────────────────────────────────┘
  ✓ ★异步下每个连接利用率高，pool_size 5~10 通常够★
  ✓ 或上 ★PgBouncer★

★ ★uvloop 和 httptools★：
  uvicorn --loop uvloop --http httptools
  ★ uvloop：★基于 libuv 的事件循环，比标准库快 2~4 倍★
  ★ httptools：★C 实现的 HTTP 解析器★
  ★ ✓ ★只对"框架层开销"有效★
  ★ ✗ ★如果瓶颈是数据库，提升几乎感知不到★
  ★ 安装 uvicorn[standard] 会自动带上

★ ★外部调用必须配超时和连接池★：
  # ✗ 每次新建客户端（★TCP 握手 + TLS 握手，很贵★）
  async def call():
      async with httpx.AsyncClient() as c:      # ★每次都建连接池★
          return await c.get(url)
  # ✓ ★复用全局客户端★
  @asynccontextmanager
  async def lifespan(app):
      app.state.http = httpx.AsyncClient(
          timeout=httpx.Timeout(5.0, connect=2.0),      # ★★必须设★★
          limits=httpx.Limits(max_connections=100,
                              max_keepalive_connections=20))
      yield
      await app.state.http.aclose()
  ★ ★没有超时的外部调用 = 对方卡住时你的连接全部堆积★

★ ★并发控制★：
  # ★限制同时进行的重操作★
  sem = asyncio.Semaphore(20)
  async def heavy():
      async with sem: ...
  # ★批量并发（有界）★
  async def fetch_all(urls):
      sem = asyncio.Semaphore(10)
      async def one(u):
          async with sem:
              return await client.get(u)
      return await asyncio.gather(*[one(u) for u in urls])
  ★ ✗ ★裸 gather 是无界并发★（1000 个 URL 就是 1000 个并发）

★ ★CPU 密集的处理★：
  ★ 线程池救不了（GIL）→ ★进程池或独立服务★
  from concurrent.futures import ProcessPoolExecutor
  pool = ProcessPoolExecutor(max_workers=4)
  result = await loop.run_in_executor(pool, cpu_heavy, arg)
  ✓ ★更好：拆成独立的 worker 服务（Celery / 专用推理服务）★

★ ★gunicorn 的实用参数★：
  --max-requests 10000 --max-requests-jitter 1000   # ★★兜住内存泄漏★★
  --timeout 60                                       # ★大于最慢的请求★
  --graceful-timeout 30                              # ★优雅退出★
  --preload                                          # ★省内存但不能热重载★

连接数的算术是最容易爆的地方：workers × (pool_size + max_overflow) 很容易超过 PostgreSQL 默认的 100——异步下每个连接的利用率高，pool_size 设 5~10 通常就够。uvloop 和 httptools 能让框架层快 2~4 倍，但如果瓶颈是数据库就几乎感知不到。外部调用有两个必做的：复用全局的 httpx.AsyncClient（每次新建要付 TCP 加 TLS 握手的代价）和必须设超时（没有超时的外部调用，对方卡住时你的连接会全部堆积）。并发要有界——asyncio.gather 裸用是无界的（1000 个 URL 就是 1000 个并发），要用 Semaphore 限制。CPU 密集线程池救不了（GIL），要用进程池或拆成独立服务。gunicorn 记得配 --max-requests 兜住内存泄漏。

五、常见性能陷阱

★ ★陷阱一：依赖里的重活（★很隐蔽★）★
  async def get_current_user(token: str, db: DbDep):
      user = await db.get(User, uid)
      ★perms = await db.scalars(select(Permission)...)★   # ★★每个请求都查★★
      return user
  ★ 影响：★所有需要认证的接口都多一次查询★
  ✓ ★把权限缓存到 Redis 或 token 里★
  ✓ ★或用 lru_cache 缓存不常变的部分★

★ ★陷阱二：中间件里的重活★
  @app.middleware("http")
  async def audit(request, call_next):
      ★await db.execute(insert(AuditLog)...)★             # ★★每个请求写库★★
      return await call_next(request)
  ✓ ★异步写入队列，批量落库★
  ✓ 或只记日志，离线分析

★ ★陷阱三：Pydantic 模型定义在函数内★
  async def handler():
      class Item(BaseModel): ...       # ★★每次调用都重新构建 core schema★★
  ★ ★V2 的模型构建是有成本的★
  ✓ 定义在模块级

★ ★陷阱四：过度使用 response_model 嵌套★
  class OrderOut(BaseModel):
      items: list[ItemOut]
      user: UserOut
      coupons: list[CouponOut]
  ★ ★深度嵌套 = 递归校验 + 递归序列化★
  ✓ ★列表接口用扁平的精简模型，详情接口才用完整嵌套★

★ ★陷阱五：同步库的隐蔽使用★
  ★ 检查清单：
    □ ★requests → httpx.AsyncClient★
    □ ★redis → redis.asyncio★
    □ ★pymongo → motor★
    □ ★boto3 → aioboto3（或 run_in_threadpool）★
    □ ★同步 SQLAlchemy Session → AsyncSession（或用 def 路由）★
    □ ★bcrypt / argon2 → run_in_threadpool★
    □ ★PIL / opencv → run_in_threadpool 或进程池★
    □ ★open() 读大文件 → aiofiles 或线程池★

★ ★陷阱六：日志开销★
  logger.debug(f"data={json.dumps(big_obj)}")     # ★★f-string 总是先求值★★
  ✓ logger.debug("data=%s", big_obj)              # ★级别不够时不格式化★
  ★ 生产环境把 sqlalchemy.engine 设 WARNING（★INFO 会打印每条 SQL★）

★ ★陷阱七：启动慢★
  ★ 原因：★Pydantic V2 在 import 时构建 core schema★
    + 大量路由的 OpenAPI schema 生成
  ★ 影响：★serverless 冷启动、k8s 滚动更新变慢★
  ✓ 延迟导入不常用的模块
  ✓ ★生产关闭 /docs 时 OpenAPI schema 仍会在首次访问时生成★
    → 可以 app.openapi_schema = None 并按需生成

★ ★陷阱八：无界的查询★
  @app.get("/items")
  async def items(limit: int = 100):     # ★★没有上限！★★
      ...limit(limit)
  ✓ limit: int = Query(100, ★le=1000★)   # ★★必须设上限★★

八个陷阱里，依赖和中间件里的重活最隐蔽——它们对每个请求都执行，一次多余的查询会放大到全站。Pydantic 模型要定义在模块级（V2 的 schema 构建有成本）。深度嵌套的 response_model 会递归校验和序列化——列表接口用扁平精简模型、详情接口才用完整嵌套。同步库的隐蔽使用那份清单值得逐条核对。日志要用 %s 惰性格式化（f-string 总是先求值），生产环境 sqlalchemy.engine 设 WARNING。启动慢主要来自 Pydantic V2 的 schema 构建，影响 serverless 冷启动。最后查询必须有上限（Query(100, le=1000)）。

六、实践清单

★ ★优化顺序（严格按这个来）★：
  ① ★测量★（找到真正的瓶颈）
  ② ★排除阻塞★（async def 里的同步调用）
  ③ ★消灭 N+1★（预加载）
  ④ ★加索引★
  ⑤ ★减少数据量★（分页、只查需要的列）
  ⑥ ★序列化优化★（ORJSONResponse、跳过 response_model）
  ⑦ ★外部调用：复用客户端 + 超时 + 并发控制★
  ⑧ ★HTTP 缓存★
  ⑨ ★Redis 缓存★
  ⑩ ★调 worker / 连接池 / uvloop★
  ⑪ 加机器

★ 检查清单：
  【阻塞】
  □ ★async def 里没有 requests/同步 DB/bcrypt/PIL★
  □ ★Redis 用 redis.asyncio★
  □ ★开发环境开 asyncio debug★
  【数据库】
  □ ★预加载（selectinload/joinedload）★
  □ ★只查需要的列★
  □ ★分页 size 有上限★
  □ ★连接数算过（workers × pool < max_connections）★
  【序列化】
  □ ★ORJSONResponse★
  □ ★大列表考虑跳过 response_model★
  □ ★模型定义在模块级★
  □ ★列表用精简模型，别深度嵌套★
  【缓存】
  □ ★key 包含用户 ID★
  □ ★带用户数据的响应标 private/no-store★
  □ ★lru_cache 不装饰 async 函数★
  □ ★TTL 有随机抖动★
  【外部调用】
  □ ★复用全局 httpx.AsyncClient★
  □ ★所有调用都有 timeout★
  □ ★gather 用 Semaphore 限并发★
  【部署】
  □ ★uvloop + httptools★
  □ ★gunicorn --max-requests 兜内存泄漏★
  □ ★--timeout 大于最慢请求★

★ ★"QPS 上不去"排查树★：
  CPU 满了吗？
   ├─ ★满了★ → CPU 密集
   │   ├─ ★序列化/加密/图片处理★ → 优化或丢进程池
   │   └─ ★纯业务计算★ → 加 worker / 加机器
   └─ ★没满★（★更常见★）
       ├─ ★asyncio debug 有慢回调告警★ → ★阻塞★
       ├─ ★线程池 token 耗尽★ → def 路由太多/太慢
       ├─ ★数据库连接池耗尽★ → 调 pool 或查慢 SQL
       ├─ ★下游服务慢★ → 加超时/熔断/缓存
       └─ ★锁竞争★（Redis 分布式锁、数据库行锁）

★ 一句话总结：
  ★"FastAPI 的瓶颈几乎从不在框架：先排除 async def 里的阻塞
    （CPU 闲但慢就是它），再消灭 N+1 和加索引，然后减少数据量；
    序列化用 ORJSONResponse（一行快 3~5 倍）、大列表可直接返回
    Response 跳过 response_model；缓存排在最后，
    且 key 必须含用户 ID、Redis 必须用异步客户端。"★

优化顺序要严格遵守——排除阻塞排在第二位（仅次于测量），因为它是 FastAPI 特有且最容易被忽略的。「QPS 上不去」的排查树很实用：先看 CPU 满不满——满了是 CPU 密集（优化或加机器），没满则更常见，依次查阻塞、线程池 token、数据库连接池、下游服务、锁竞争。检查清单里 FastAPI 特有的三条：Redis 用 redis.asyncio、lru_cache 不装饰 async 函数、大列表考虑跳过 response_model。

记忆钩子：「★FastAPI 号称高性能，但真实项目的瓶颈几乎从不在框架本身★（ASGI+路由 <0.1ms、依赖解析和校验各 1ms 上下）。按频率排：★① 数据库 N+1 和慢查询 ② 在 async def 里写了阻塞代码 ③ 序列化开销 ④ 同步外部调用没超时 ⑤ worker/连接池配置★。★优化顺序：测量 → 排除阻塞 → 消灭 N+1 → 加索引 → 减少数据量 → 序列化优化 → 外部调用 → HTTP 缓存 → Redis 缓存 → 调 worker，缓存排在很后面★。★第一件事是排除阻塞（FastAPI 特有且最易忽略）★——症状是★『CPU 占用很低但 QPS 上不去、P99 随并发线性恶化』★，检测用 ★asyncio debug 的 slow_callback_duration=0.1★、blockbuster、py-spy dump、★或压测对比并发 1 和 50 的 P99（正好差 50 倍就是完全串行）★。★序列化是 FastAPI 特有的性能点★：★response_model 不只筛字段，它会『用模型重新校验一遍 + jsonable_encoder 递归转换 + json.dumps』，1000 条记录就各跑 1000 次★；两个优化：★ORJSONResponse 一行快 35 倍★（Rust 实现、原生支持 datetime/UUID）、★大列表直接返回 Response 实例可完全跳过序列化链★（FastAPI 检测到返回值是 Response 就直接用，★代价是失去字段过滤和文档，要自己保证不泄露敏感字段★）；★但算例显示分页才是根本★（5000 条从 450ms 优化到 35ms 靠跳过 model+只查列，分页直接到 2ms）。★缓存四层：请求内（依赖缓存本身就是）、lru_cache（只适合不变数据，多 worker 每进程一份，★且不能装饰 async 函数——它缓存 coroutine 对象，第二次 await 会报错★）、Redis、HTTP（收益最大因为请求根本不到应用）★。三个缓存红线：★① 必须用 redis.asyncio 而不是同步客户端★（否则阻塞事件循环，加缓存反而更慢）、★② key 必须包含所有影响响应的因素尤其是用户 ID（漏了就是数据串号）★、★③ 带用户数据的响应必须标 private/no-store（否则 CDN 可能把 A 的数据发给 B）★；★fastapi-cache2 的默认 key 含 URL 但不含用户，必须自定义 key_builder★。★连接数算术容易爆：workers × (pool_size + max_overflow) 常超过 PG 默认的 100，异步下 pool_size 510 就够★。外部调用两个必做：★复用全局 httpx.AsyncClient（每次新建要付 TCP+TLS 握手）★ + ★必须设 timeout（没超时时对方卡住你的连接全堆积）★，★gather 要用 Semaphore 限并发（裸 gather 是无界的）★。隐蔽陷阱：★依赖和中间件里的重活会放大到每个请求★、★Pydantic 模型要定义在模块级★、★日志用 %s 惰性格式化★、★查询必须有上限 Query(100, le=1000)★。」

七、常见误区与追问

• 误区：接口慢就加 Redis 缓存。 缓存应该排在优化顺序的倒数第三位，前面还有更划算的手段。它引入的复杂度很实在：失效逻辑、一致性窗口、Redis 的可用性和运维、序列化开销、以及最危险的缓存串号（key 设计漏了用户 ID 会把 A 的数据返给 B）。而且很多时候它根本没解决问题——如果慢的原因是一条扫全表的 SQL，加了缓存只是让「第一个请求」和「缓存过期后的那个请求」继续等，P99 依然很糟；如果慢的原因是 async def 里有阻塞调用，那缓存命中率再高也救不了被卡死的事件循环。正确顺序是：测量 → 排除阻塞 → 消灭 N+1 → 加索引 → 减少数据量（分页、只查需要的列）→ 序列化优化 → 外部调用优化 → HTTP 缓存 → 才轮到 Redis。而且做完前面这些之后，你会更清楚「到底该缓存什么」。
• 误区：response_model 只是声明返回格式，运行时没什么开销。 它是实打实的三步开销：① 用这个模型重新校验你返回的数据（不是简单的字段筛选，而是完整的 Pydantic 校验流程）；② jsonable_encoder 递归遍历整个对象树，把 datetime、Decimal、UUID、Enum、嵌套的 BaseModel 转成 JSON 兼容类型；③ json.dumps。返回 1000 条记录时，这三步各跑 1000 次；如果模型还有嵌套（OrderOut 里含 items: list[ItemOut]），开销还要乘上嵌套层数。两个优化手段：换 ORJSONResponse（app = FastAPI(default_response_class=ORJSONResponse)，一行，序列化快 3~5 倍）；大列表接口直接返回 Response 实例——FastAPI 在拿到返回值时会先判断 isinstance(raw_response, Response)，是的话直接使用、完全跳过上面三步。代价是失去了字段过滤（要自己保证不泄露 hashed_password 这类字段）和自动文档（可以用 responses= 参数手动补）。
• 误区：加了缓存就一定变快。 用错客户端反而更慢。最典型的是在异步应用里用了同步的 Redis 客户端：import redis 后 r.get(key) 是阻塞调用——在 async def 里执行它会卡住整个事件循环，所有并发请求排队等待。结果就是：本来查数据库 10ms，加了缓存变成「查 Redis 1ms 但阻塞了事件循环」，并发一上来整体吞吐反而下降。正确的是 import redis.asyncio as aioredis。同样的问题也出现在 HTTP 调用（requests vs httpx.AsyncClient）、文件读写、以及第三方 SDK（boto3、stripe 内部都是 requests）。另一个「加了缓存更慢」的场景是缓存了大对象：如果缓存的值有几 MB，pickle/json 的序列化反序列化加上 Redis 的网络传输，可能比直接查数据库还慢——这时候缓存就是纯粹的负收益。
• 误区：@lru_cache 可以用来缓存异步函数的结果。 不能，而且会产生诡异的错误。lru_cache 缓存的是函数的返回值——而 async def 函数被调用时返回的是一个 coroutine 对象（还没执行）。所以第一次调用时，lru_cache 把这个 coroutine 存了起来，你 await 它拿到结果；第二次调用命中缓存，返回的是同一个已经被 await 过的 coroutine 对象——再次 await 会抛 RuntimeError: cannot reuse already awaited coroutine。解决方案有三个：① 用 async-lru 库的 @alru_cache（专门为协程设计）；② 自己实现——用一个 dict 缓存结果值（不是 coroutine），并配合 asyncio.Lock 防止并发时重复计算（单飞）；③ 把「纯计算部分」抽成同步函数再用 lru_cache。顺带提醒 lru_cache 的另外两个限制：多 worker 下每个进程一份（数据变了各进程不一致）、没有 TTL（可以用「时间桶」技巧：把 int(time.time()) // 60 作为一个参数传进去，每分钟自然换 key）。
• 误区：把 worker 数调大就能提升吞吐。 要看瓶颈在哪，而且会撞上连接数上限。如果瓶颈是 CPU（序列化、加密、图片处理），加 worker 确实有效（多进程绕过 GIL）；但如果瓶颈是数据库慢查询或下游服务，加 worker 只会让更多请求同时去挤那个瓶颈，P99 反而更差。而且有一个硬约束：总连接需求 = worker 数 × (数据库 pool_size + max_overflow)——4 个 worker 配 pool_size=10, max_overflow=20 就是 120 个连接，超过 PostgreSQL 默认的 max_connections=100，表现是「部分请求报 too many connections」。异步应用的一个特点是每个连接的利用率很高（不像同步那样一个请求独占一个连接很久），所以 pool_size 设 5~10 通常就够；连接数确实紧张时上 PgBouncer。经验配置是 worker 数 ≈ CPU 核数，然后根据实测的 P99 和连接池使用率微调。
• 追问：ORJSONResponse 值得全局启用吗？ 值得，几乎没有副作用。orjson 是 Rust 实现的 JSON 库，序列化速度是标准库的 3~5 倍，而且原生支持 datetime、date、UUID、dataclass、numpy 数组（标准库需要自定义 encoder）。启用方式是一行：app = FastAPI(default_response_class=ORJSONResponse)，之后所有默认返回都走它。需要注意的差异有三点：① 输出更紧凑——orjson 默认不在分隔符后加空格（{"a":1} 而不是 {"a": 1}），响应体略小，但如果有测试断言了精确的 JSON 字符串会失败；② 对某些类型的处理不同——比如它默认把 datetime 序列化成 RFC 3339 格式，而且不支持某些自定义类型（需要传 default= 参数）；③ 是 C 扩展——某些受限环境（无法编译、特殊 CPU 架构）可能装不上。实践上：新项目直接全局开启；已有项目开启后跑一遍测试看有没有断言精确 JSON 字符串的地方。另一个选择是 UJSONResponse，但 ujson 在数值精度和边界情况上不如 orjson 严谨。
• 追问：缓存 key 怎么设计才不会串数据？ 核心原则是**「key 必须包含所有会影响响应内容的因素」。最常被漏掉的是用户身份**——@cache(expire=60) 装饰一个 /my/orders 接口，默认 key 只包含 URL，结果就是第一个访问的用户的订单被缓存下来发给了所有人，这是最严重的一类生产事故（数据泄露）。完整的 key 应该包含：① 版本前缀（v2:，便于全量失效）；② 用户 ID 或租户 ID（任何用户相关的数据）；③ 全部查询参数（分页、筛选、排序——漏了分页参数会导致第 2 页显示第 1 页的内容）；④ 语言/时区（如果响应会因此不同）；⑤ 权限维度（管理员和普通用户看到的字段不同时）。实践建议：写一个统一的 key_builder 函数，强制传入 user_id，而不是每个接口各自拼字符串。另外还有一个相关的红线：HTTP 层面，带用户数据的响应必须设 Cache-Control: private 或 no-store——否则 CDN 或中间代理可能缓存下来发给别人，后果和缓存串号一样严重。
• 追问：怎么系统地排查「QPS 上不去」？ 先看一个关键指标：CPU 使用率。① CPU 接近饱和——说明是 CPU 密集型瓶颈，往下细分：是序列化/加密/图片处理这类可优化的（换 orjson、丢进程池、异步化），还是纯业务计算（那就加 worker 或加机器）。② CPU 很闲但 QPS 上不去——这是更常见也更有价值的情况，说明有东西在「等」或者被「卡住」，依次排查：（a）事件循环阻塞——开 asyncio debug 看有没有慢回调告警，用 py-spy dump 看线程栈是不是卡在 requests、bcrypt、同步数据库调用上；（b）线程池 token 耗尽——def 路由太多或太慢，40 个 token 打满后新请求排队，检查 anyio.to_thread.current_default_thread_limiter().available_tokens；（c）数据库连接池耗尽——所有连接都在等慢查询，看连接池的 checkout 时间和数据库的活跃会话；（d）下游服务变慢——没配超时的外部调用会让请求堆积；（e）锁竞争——Redis 分布式锁、数据库行锁、或者代码里的 asyncio.Lock 粒度太粗。这套排查树能覆盖绝大多数情况。

八、加强记忆

FastAPI 号称高性能，但真实项目的瓶颈几乎从不在框架本身（ASGI 加路由 < 0.1ms、依赖解析和校验各 1ms 上下）。按频率排：① 数据库 N+1 和慢查询 ② 在 async def 里写了阻塞代码 ③ 序列化开销 ④ 同步外部调用没设超时 ⑤ worker/连接池配置不合理。优化顺序是：测量 → 排除阻塞 → 消灭 N+1 → 加索引 → 减少数据量 → 序列化优化 → 外部调用 → HTTP 缓存 → Redis 缓存 → 调 worker，缓存排在很后面。第一件事是排除阻塞（FastAPI 特有且最容易被忽略）——症状是**「CPU 占用很低但 QPS 上不去、P99 随并发线性恶化」，检测手段有 asyncio debug 的 slow_callback_duration=0.1、blockbuster、py-spy dump、以及压测对比并发 1 和并发 50 的 P99（正好差 50 倍就是完全串行）。序列化是 FastAPI 特有的性能点：response_model 不只是筛选字段，它会「用模型重新校验一遍 + jsonable_encoder 递归转换 + json.dumps」，1000 条记录就各跑 1000 次**；两个优化是 ORJSONResponse（一行，快 3~5 倍，Rust 实现且原生支持 datetime/UUID） 和大列表直接返回 Response 实例完全跳过序列化链（FastAPI 检测到返回值是 Response 就直接用，代价是失去字段过滤和文档，要自己保证不泄露敏感字段）；但算例显示分页才是根本（5000 条从 450ms 优化到 35ms 靠的是跳过 model 加只查列，而分页直接降到 2ms）。缓存分四层：请求内（依赖缓存本身就是）、lru_cache（只适合不变的数据，多 worker 下每进程一份，而且不能装饰 async 函数——它缓存的是 coroutine 对象，第二次 await 会报错）、Redis、HTTP（收益最大，因为请求根本到不了应用）。三条缓存红线：① 必须用 redis.asyncio 而不是同步客户端（否则阻塞事件循环，加缓存反而更慢）、② key 必须包含所有影响响应的因素，尤其是用户 ID（漏了就是数据串号）、③ 带用户数据的响应必须标 private/no-store（否则 CDN 可能把 A 的数据发给 B）；注意 fastapi-cache2 的默认 key 含 URL 但不含用户，必须自定义 key_builder。连接数的算术容易爆：workers × (pool_size + max_overflow) 常超过 PG 默认的 100，异步下 pool_size 设 5~10 就够。外部调用有两个必做的：复用全局 httpx.AsyncClient（每次新建要付 TCP + TLS 握手的代价）和必须设 timeout（没超时时对方卡住会让你的连接全部堆积），gather 要用 Semaphore 限并发（裸 gather 是无界的）。其他隐蔽陷阱：依赖和中间件里的重活会放大到每个请求、Pydantic 模型要定义在模块级、日志用 %s 惰性格式化、查询必须有上限 Query(100, le=1000)。