FastAPI性能调优实录：Profiling定位与多级缓存策略

一、问题背景：一个“看起来没毛病”的接口

事情是这样的：上周四下午，运维同事在群里扔了个截图，说某个核心聚合接口的P99延迟已经飘到2秒以上，而且CPU使用率从30%一路飙到85%。我第一反应是“是不是数据库慢查询”，结果看了慢日志，只有几条普通的索引扫描，最慢的也就400ms。

这个接口做的事情很简单：根据用户ID返回其最近30天的订单列表，附带商品快照、店铺信息和优惠券状态。逻辑不复杂，表也不大（订单表约200万行），但就是慢。更诡异的是，同样的逻辑用PostgreSQL的psql客户端执行，SQL总耗时不超过50ms。

环境与版本：
- Python 3.11.4 / FastAPI 0.104.1 / Uvicorn 0.24.0（worker=2，--limit-concurrency 1024）
- SQLAlchemy 2.0.21 + asyncpg 0.28
- PostgreSQL 15.3（16核32G，SSD，shared_buffers=4GB）
- Redis 7.0（单机，maxmemory 2GB）
- 压测工具：wrk 4.2.0（单机，500并发，60秒）

二、Profiling第一轮：cProfile没找到真凶

我习惯先用cProfile跑一次单请求，因为快且不需要额外部署。写了个测试脚本调用接口，然后python -m cProfile -s cumtime。结果出乎意料，消耗最大的既不是数据库查询，也不是业务逻辑，而是pydantic的序列化——BaseModel.dict()占用了总耗时的42%。

# 复现问题的简化代码
from pydantic import BaseModel

class OrderItem(BaseModel):
    order_id: str
    sku_id: str
    amount: float
    status: int
    # 还有10个字段...

class OrderListResponse(BaseModel):
    items: list[OrderItem]
    total: int
    page: int

当时每个订单会嵌套商品快照（又是一个dict），一个订单列表平均有80个订单，每个订单带3个商品，就是240个嵌套对象。Pydantic在构建这些对象时，由于字段类型验证和from_orm的转换，耗时呈指数级增长。

结论：先别急着优化SQL，先看看数据在Python内存里浪费了多少时间。

三、Profiling第二轮：py-spy定位真实瓶颈

cProfile只能看到函数级统计，但无法告诉你等待I/O的时间。我换用py-spy dump --pid 连续抓了20次栈，发现大量线程阻塞在asyncpg的Connection.execute上。但奇怪的是，SQL本身很快——问题在于查询次数。

继续追，发现代码里有个经典的N+1陷阱：

# 错误示例：循环内查询
orders = await session.execute(
    select(Order).where(Order.user_id == user_id).limit(80)
)
order_items = []
for order in orders.scalars().all():
    # 每个订单单独查询商品
    skus = await session.execute(
        select(Sku).where(Sku.id.in_(order.sku_ids))
    )
    # 每个订单单独查询店铺
    shop = await session.execute(
        select(Shop).where(Shop.id == order.shop_id)
    )
    order_items.append(...)

80个订单 = 80次商品查询 + 80次店铺查询 + 80次优惠券查询 = 240次往返。虽然单次只要1ms，但串行下来就是240ms。加上Pydantic的序列化，整体就到了1200ms。

优化方案：
1. 用SQLAlchemy 2.0的selectinload一次性预加载所有关联对象。
2. 把响应模型改成response_model_exclude_none=True，减少序列化字段。
3. 引入Redis缓存整个聚合结果（key = user:{id}:orders:30d，TTL 5分钟）。

四、核心实现：预加载 + 缓存 + 压缩

4.1 SQLAlchemy预加载

from sqlalchemy.orm import selectinload
from sqlalchemy.ext.asyncio import AsyncSession

async def get_orders_with_details(session: AsyncSession, user_id: int):
    # 一次性加载订单+商品+店铺，消除N+1
    stmt = (
        select(Order)
        .options(
            selectinload(Order.skus),
            selectinload(Order.shop),
            selectinload(Order.coupon),
        )
        .where(Order.user_id == user_id)
        .order_by(Order.created_at.desc())
        .limit(80)
    )
    result = await session.execute(stmt)
    return result.scalars().unique().all()

注意：必须加.unique()，否则由于selectinload产生重复行。

4.2 Redis二级缓存

import json
import aioredis
from fastapi import Request

APP_CACHE = aioredis.from_url("redis://localhost:6379/0", decode_responses=True)

async def get_orders_cached(request: Request, user_id: int):
    cache_key = f"user:{user_id}:orders:30d"
    cached = await APP_CACHE.get(cache_key)
    if cached:
        return json.loads(cached)  # 直接返回，省去DB和序列化

    orders = await get_orders_with_details(request.state.db, user_id)
    response = [order_to_dict(o) for o in orders]  # 手动转dict，跳过Pydantic
    await APP_CACHE.setex(cache_key, 300, json.dumps(response))
    return response

踩坑：直接用json.dumps比Pydantic快5倍，因为Pydantic的dict()对嵌套模型会递归调用验证。这里我手写一个轻量转换函数，只提取前端需要的字段。

4.3 gzip中间件

在FastAPI中开启gzip压缩，尤其响应体超过1KB时收益明显：

from starlette.middleware.gzip import GZipMiddleware

app = FastAPI()
app.add_middleware(GZipMiddleware, minimum_size=1024, compresslevel=5)

压缩级别5是权衡CPU和体积的最佳点。实测响应体从12KB压缩到3.2KB，网络传输时间下降70%。

五、压测数据：从1200ms到180ms

使用wrk压测60秒，500并发：

需要说明的是：缓存命中率在压测场景下是100%（同一user_id反复请求）。线上实际命中率约78%，因为用户分布较分散。命中时延迟约15ms（Redis读+反序列化），未命中时约220ms（DB查询+组装）。

额外优化：
- 把limit(80)改成limit(30)，前端实际只需要展示30条，减少序列化对象数量。
- 将Pydantic的response_model改为ORJSONResponse（orjson库），序列化速度再快20%。

六、踩坑与反思

坑1：selectinload与limit的相互作用
SQLAlchemy的selectinload会先查主表再IN查询关联表，如果主表用了limit，预加载不会生效（因为无法在子查询中复用LIMIT）。解决方案是：先查询主表ID列表，再用where_in加载关联表。但这里我们limit 80，selectinload是生效的，因为SQLAlchemy会先执行子查询获取ID再加载。

坑2：Redis缓存穿透
如果用户没有订单，会缓存空列表，但TTL设短一点（60秒），避免缓存雪崩。

坑3：py-spy在容器里需要特权
如果使用Docker容器，记得加--pid=host并设置SYS_PTRACE能力，否则无法attach到uvicorn进程。

总结：这次的调优思路是“先Profile，再动手”。千万不要凭直觉去优化数据库索引——我们数据库本身没问题，问题在应用层的查询次数和序列化开销。
如果你也在用FastAPI或Flask遇到类似问题，建议先跑一次py-spy dump看看线程栈，80%的概率能找到惊喜。

后续迭代方向：
- 将缓存预热到本地内存（cachetools），减少一次Redis RTT。
- 使用asyncpg原生连接池替代SQLAlchemy的池，再降10%延迟。
- 如果QPS继续上涨，考虑引入消息队列异步更新缓存，而不是直接失效。

代码已上传到GitHub：github.com/yourname/fastapi-perf-tuning（含wrk脚本和py-spy分析脚本）。有问题欢迎评论区讨论。