FastAPI+Flask双栈API性能调优：从1800ms到95ms的profiling、SQL与缓存改造实录

一、问题背景：CPU 100%，数据库却很闲

事情起因是监控告警。我们有一个商品聚合查询接口 GET /api/v1/products/{id}/detail，返回商品基础信息、SKU 列表、库存、最近 20 条评价摘要。日请求量约 200 万，高峰期 QPS 在 350 左右。

现象很怪：

• 接口 P95 延迟 1800ms，P99 冲到 3.2s
• 服务 Pod 的 CPU 使用率稳定在 95% 以上，8 核机器
• 但 RDS 的 CPU 只有 20%，慢查询日志里几乎没有记录
• 单条请求平均返回 JSON 约 45KB

第一反应是"数据库慢"，但数据不支持这个结论。那就老老实实做 profiling，别猜。

二、环境与版本

先把环境交代清楚，不然数据没法复现：

机器规格：API 服务 8C16G × 3 副本，压测机 16C32G 独立部署，避免互相干扰。

三、方案设计：先定位，再动手

我的原则是：没有 profiling 数据的优化都是玄学。整体分三步走：

1. 定位阶段：py-spy 采样火焰图看 CPU 花在哪，cProfile 看函数级耗时，EXPLAIN 看 SQL 执行计划
2. 优化阶段：针对热点分别处理——ORM 查询改成批量预加载、加复合索引、引入两级缓存
3. 验证阶段：wrk 固定并发压测，对比 P50/P95/P99 和 QPS，同时观察 CPU 与 DB 指标

这里有个关键判断：CPU 高但 DB 闲，说明瓶颈大概率在应用层的 CPU 密集操作（序列化、循环、对象构造），而不是等待 IO。后面 profiling 证实了这个猜测。

四、核心实现

4.1 用 py-spy 抓火焰图

线上服务不能随便挂 profiler，py-spy 的好处是 attach 到运行中的进程，几乎零开销：

# 找到 worker 进程
ps -ef | grep uvicorn

# 采样 30 秒，生成火焰图
py-spy record -o flame.svg --pid 12345 --duration 30 --rate 100

# 或者直接看 top 函数
py-spy top --pid 12345

火焰图一出来，问题非常明显：超过 40% 的 CPU 时间花在 pydantic 的 model_validate 和 jsonable_encoder 上，另外 25% 花在 SQLAlchemy 的 Session.get 循环调用里——典型的 N+1。

4.2 定位 N+1 查询

原代码长这样（简化版）：

# 问题代码：N+1 + 重复序列化
@app.get("/api/v1/products/{pid}/detail")
def get_detail(pid: int, db: Session = Depends(get_db)):
    product = db.get(Product, pid)
    skus = db.query(Sku).filter(Sku.product_id == pid).all()      # 1 次
    for sku in skus:
        sku.stock = db.query(Stock).filter(Stock.sku_id == sku.id).first()  # N 次
    reviews = db.query(Review).filter(Review.product_id == pid)\
                .order_by(Review.created_at.desc()).limit(20).all()
    return {"product": product, "skus": skus, "reviews": reviews}

一个商品平均 12 个 SKU，等于每次请求打 1 + 1 + 12 + 1 = 15 条 SQL。虽然单条都很快（走主键/索引），但连接池争用和 ORM 对象构造开销叠加起来非常可观。

改成 selectinload 批量预加载，SQL 数量从 15 降到 3：

from sqlalchemy.orm import selectinload
from sqlalchemy import select

@app.get("/api/v1/products/{pid}/detail")
def get_detail(pid: int, db: Session = Depends(get_db)):
    stmt = (
        select(Product)
        .options(
            selectinload(Product.skus).selectinload(Sku.stock),
            selectinload(Product.reviews),
        )
        .where(Product.id == pid)
    )
    product = db.execute(stmt).scalar_one_or_none()
    if product is None:
        raise HTTPException(status_code=404)
    return build_detail_response(product)

selectinload 会发一条 WHERE sku_id IN (...) 的批量查询，而不是逐条 get。这里别用 joinedload——一对多 join 会导致结果集笛卡尔膨胀，12 个 SKU × 20 条评价 = 240 行，反而更慢。一对多场景 selectinload 是更稳的选择。

4.3 索引优化

对评价表按 (product_id, created_at DESC) 建复合索引，避免 filesort：

ALTER TABLE review ADD INDEX idx_product_created (product_id, created_at DESC);

用 EXPLAIN ANALYZE 验证前后：

-- 优化前：type=ref, rows=1240, Extra=Using filesort
-- 优化后：type=ref, rows=20, Extra=Using index condition
EXPLAIN ANALYZE
SELECT * FROM review WHERE product_id = 10086
ORDER BY created_at DESC LIMIT 20;

实测该查询从 8.2ms 降到 0.6ms。注意 MySQL 8.0 才支持降序索引，5.7 上写 DESC 会被忽略，这点要留意。

4.4 序列化优化：绕开 pydantic 的重复校验

原来返回 ORM 对象，FastAPI 内部会走一遍 jsonable_encoder + pydantic 校验。数据量 45KB 的时候，这部分开销很吓人。改成在 ORM 层直接构造 dict，用 ORJSONResponse 输出：

from fastapi.responses import ORJSONResponse

def build_detail_response(product) -> dict:
    return {
        "id": product.id,
        "name": product.name,
        "skus": [
            {"id": s.id, "price": str(s.price), "stock": s.stock.qty if s.stock else 0}
            for s in product.skus
        ],
        "reviews": [
            {"id": r.id, "score": r.score, "content": r.content}
            for r in product.reviews[:20]
        ],
    }

@app.get("/api/v1/products/{pid}/detail", response_class=ORJSONResponse)
def get_detail(pid: int, db: Session = Depends(get_db)):
    ...

ORJSONResponse 比默认的 JSONResponse 快 2~3 倍，尤其在大 payload 场景。代价是失去 pydantic 的响应模型校验——我们通过单元测试覆盖字段结构来补这个缺口，可以接受。

4.5 两级缓存

商品详情是典型的读多写少，天然适合缓存。设计成两级：

• L1：进程内 LRU（cachetools.TTLCache，maxsize=2000，ttl=30s），挡住突发重复请求
• L2：Redis 7.2，TTL 300s，带随机抖动（±30s）防雪崩

import json, random
from cachetools import TTLCache
import redis

l1 = TTLCache(maxsize=2000, ttl=30)
rds = redis.Redis(host="redis.internal", port=6379, decode_responses=True)

def get_cached_detail(pid: int, db: Session):
    key = f"prod:detail:{pid}"
    if (hit := l1.get(key)) is not None:
        return hit
    if (raw := rds.get(key)) is not None:
        data = json.loads(raw)
        l1[key] = data
        return data
    data = build_detail_response(load_product(db, pid))
    ttl = 300 + random.randint(-30, 30)
    rds.setex(key, ttl, json.dumps(data, ensure_ascii=False))
    l1[key] = data
    return data

Flask 服务那边结构一样，只是把 FastAPI 的依赖注入换成 before_request 里的连接管理。这里不重复贴代码。

4.6 踩坑记录

坑一：缓存 key 没带版本号。 上线新字段后，老缓存还在返回旧结构，前端直接报错。后来 key 统一加 schema 版本前缀 prod:detail:v2:{pid}，发布时 bump 版本号即可。

坑二：gevent + Redis 连接池。 Flask 服务用 gevent worker，最初每个请求新建 Redis 连接，压测时出现 Too many open files。改成模块级单例连接池 max_connections=50 后解决。

坑三：selectinload 的 IN 参数爆炸。 有个商品挂了 800 多个 SKU，IN 列表过长导致 SQL 解析变慢。加了 max_skus=200 的业务上限，超出的走异步导出。

坑四：L1 缓存和写操作不一致。 商品改价后 30s 内仍返回旧价。改成写操作时主动 l1.pop(key) + rds.delete(key)，读多写少场景下这个成本可以接受。

五、压测数据对比

压测条件：wrk 固定 200 并发连接、持续 5 分钟、请求分布按线上热点商品 ID 的 Zipf 分布模拟。每轮压测前清空缓存并预热 1 分钟。

分阶段看贡献度（FastAPI 侧，P95）：

• 仅修 N+1 + 索引：1810ms → 780ms
• 加 ORJSONResponse + 序列化改造：780ms → 410ms
• 加两级缓存：410ms → 95ms

可以看到，缓存贡献最大，但前两步是必须的——如果直接上缓存，缓存穿透时的毛刺依然会打到 700ms 以上，而且缓存失效瞬间容易击穿 DB。

Redis 侧观察：命中缓存时 RT 约 0.4ms（同机房），L1 命中约 0.02ms。Redis 内存占用约 1.2GB（200 万商品详情，平均 600B/条，含压缩）。

六、总结

这次调优最大的教训是："CPU 高 DB 闲"这个信号要第一时间重视，它几乎直接指向应用层而非存储层。如果一上来就去加数据库从库或者换连接池，纯属浪费。

几个可以直接复用的结论：

1. py-spy attach 生产进程是安全的，采样率 100Hz 对性能影响 < 2%，比停机排查划算得多
2. 一对多关系用 selectinload 而非 joinedload，避免结果集膨胀
3. MySQL 8.0 的降序复合索引对 ORDER BY ... DESC LIMIT n 效果显著
4. 大 payload 场景换 ORJSONResponse，零业务改动拿到 2 倍以上序列化提升
5. 两级缓存里 L1 的 TTL 要显著小于 L2（30s vs 300s），否则一致性问题会被放大
6. 缓存 key 一定要带 schema 版本号，这是血泪教训

还有一点没做但值得提：目前缓存是请求时懒加载，热点商品首次请求仍会穿透。下一步打算加一个基于访问计数的后台预热任务，把 Top 5000 商品的缓存提前刷好。另外考虑把评价摘要拆成独立接口，让主接口 payload 从 45KB 降到 12KB 左右，进一步压 P99。

调优这事没有终点，只有当前性价比最高的那一档。先把 P95 从 1.8s 干到 95ms，剩下的等业务量再涨一波再说。