FastAPI+Flask双栈API性能调优：从1200ms到85ms的profiling与缓存实践

一、问题背景：一个“看起来很简单”的接口

事情起因是运营反馈商品详情页偶尔转圈两秒以上。这个接口逻辑上确实不复杂：根据 product_id 返回商品基础信息、SKU 列表、库存、以及前 20 条评论。技术栈是混合的——主站网关和大部分业务在 Flask 3.0（Python 3.11）上跑，新做的商品服务用 FastAPI 0.110 独立部署，网关层用 Flask 做转发和聚合。

上线初期没人在意，日活上来之后监控开始报警：Prometheus 里 http_request_duration_seconds 的 P95 稳定在 1200ms 左右，P99 冲到 2.3s，压测 QPS 只有 180 就开始出现超时。数据库是 PostgreSQL 15，连接池用 SQLAlchemy 2.0 的默认配置。

当时第一反应是“加机器”，但直觉告诉我 180 QPS 配 4C8G 的 Pod 明显不正常，先 profiling 再说。这篇博客就是把整个定位和优化过程完整写下来，包括中间踩的坑。

二、环境与版本

先把环境列清楚，避免读者对不上号：

• Python 3.11.7
• FastAPI 0.110.0 + uvicorn 0.29.0（workers=2，默认配置）
• Flask 3.0.2 + gunicorn 21.2.0（workers=4, worker_class=gthread）
• SQLAlchemy 2.0.29（同步 ORM，pool_size=5, max_overflow=10）
• PostgreSQL 15.6，商品表 80 万行，SKU 表 320 万行，评论表 1200 万行
• Redis 7.2.4（单机，maxmemory 2gb, allkeys-lru）
• 压测工具：wrk 4.2.0 + locust 2.29.0（用来做阶梯加压）

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

我的原则是先拿到火焰图再改代码，不然容易“优化了个寂寞”。整体分三步：

第一步，profiling 定位。 FastAPI 服务用 py-spy record 做采样，Flask 网关用 py-spy top 看实时热点，同时对 SQL 层开启 echo 日志统计慢查询。压测用 wrk 固定 200 并发打 60 秒，保证采样窗口足够。

第二步，按热点分类治理。 从火焰图能看出三类问题：ORM 的 N+1 查询（评论和 SKU 各一次）、pydantic 模型序列化开销、以及 Redis 里缓存了但 key 设计不合理导致的命中率低。

第三步，分层验证。 每改一项单独压测，记录 QPS 和 P95，避免多项改动混在一起说不清谁起作用。

四、核心实现

4.1 profiling：火焰图暴露的真相

先用 py-spy 抓 FastAPI 进程：

py-spy record -o profile.svg --pid $(pgrep -f "uvicorn app.main:app") --duration 60 --rate 200

火焰图上最宽的一条是 psycopg2 的 execute，占了 60% 以上采样，再往下展开是 selectin 相关的 ORM 加载。同时把 SQLAlchemy 的 echo=True 打开，日志里同一个请求打了 23 条 SQL——典型的 N+1。

原始代码大概长这样（简化）：

# 优化前：一次请求 23 条 SQL
@app.get("/products/{pid}")
def get_product(pid: int, db: Session = Depends(get_db)):
    product = db.get(Product, pid)              # 1 条
    skus = product.skus                          # 1 条
    for sku in skus:
        _ = sku.inventory                        # N 条
    comments = db.query(Comment).filter(
        Comment.product_id == pid
    ).order_by(Comment.created_at.desc()).limit(20).all()  # 1 条
    return {"product": product, "skus": skus, "comments": comments}

注意 product.skus 是 lazy load，sku.inventory 又是关联对象，20 个 SKU 就是 20 条额外查询。加上评论和主表，单请求 23 条 SQL 完全对得上。

4.2 数据库查询优化：selectinload + 索引

改法很直接：用 SQLAlchemy 2.0 的 selectinload 预加载，把 N+1 压成固定 3 条 SQL。

# 优化后：固定 3 条 SQL
from sqlalchemy.orm import selectinload
from sqlalchemy import select

@app.get("/products/{pid}")
def get_product(pid: int, db: Session = Depends(get_db)):
    stmt = (
        select(Product)
        .options(
            selectinload(Product.skus).selectinload(Sku.inventory),
            selectinload(Product.comments),
        )
        .where(Product.id == pid)
    )
    product = db.execute(stmt).scalar_one_or_none()
    if not product:
        raise HTTPException(404)
    return product

这里有个坑：selectinload(Product.comments) 会把全部评论捞出来，1200 万行的表里热门商品有上万条评论。所以评论必须单独查并加 limit，不能塞进预加载。最终结构是主表+SKU 用 selectinload（3 条 SQL），评论单独一条带 limit 的查询（1 条），总共 4 条。

索引方面，comments(product_id, created_at DESC) 建了复合索引，skus(product_id) 原本就有。改完单请求 SQL 从 23 条降到 4 条，DB 侧耗时从 700ms 降到 90ms 左右。

4.3 缓存策略：两级缓存 + 合理 key

DB 优化完 P95 到 400ms，还不够。这时候上缓存。设计是本地 LRU（进程内）+ Redis 两级：

• L1：cachetools.TTLCache(maxsize=10000, ttl=30)，扛住同一商品的热点重复请求
• L2：Redis，key 设计为 prod:v2:{pid}，TTL 300s，value 用 orjson 序列化（比默认 json 快 3 倍左右）
• 缓存穿透：空结果也缓存 60s，value 存 "__nil__"
• 缓存击穿：用 Redis SET NX PX 做简易互斥锁，只让一个请求回源

import orjson
from cachetools import TTLCache
from redis import Redis

local_cache = TTLCache(maxsize=10000, ttl=30)
redis_cli: Redis = Redis(host="redis", port=6379, decode_responses=False)

def get_product_cached(pid: int, db: Session):
    if (hit := local_cache.get(pid)) is not None:
        return hit
    key = f"prod:v2:{pid}"
    if (raw := redis_cli.get(key)) is not None:
        data = orjson.loads(raw)
        local_cache[pid] = data
        return data
    # 简易互斥，防击穿
    lock = redis_cli.set(f"lock:{key}", "1", nx=True, px=2000)
    if not lock:
        # 拿不到锁就短暂重试，避免直接回源
        import time; time.sleep(0.05)
        raw = redis_cli.get(key)
        return orjson.loads(raw) if raw else _load_from_db(pid, db)
    try:
        data = _load_from_db(pid, db)
        redis_cli.set(key, orjson.dumps(data), ex=300)
        local_cache[pid] = data
        return data
    finally:
        redis_cli.delete(f"lock:{key}")

这里踩了个大坑：一开始 L1 缓存没设上限，dict 直接内存泄漏，跑了 6 小时 OOM 被 kill。换成 TTLCache 后稳定。另外 value 用 orjson 序列化时，datetime 需要 orjson.OPT_SERIALIZE_DATETIME（新版默认支持），早期版本会直接抛 TypeError，这个坑我查了半小时。

4.4 网关层与 worker 调优

Flask 网关那边用的是 gthread，同步转发 FastAPI，火焰图显示大量时间在 requests 库的 socket 等待上。把转发改成 httpx 的连接池复用，并且 gunicorn 从 gthread 换成 gevent（worker_class=gevent, workers=4, worker_connections=2000），网关侧 P95 从 180ms 降到 40ms。

FastAPI 侧 uvicorn 原来 workers=2，压测发现 CPU 利用率只到 60%，调到 workers=4（与 vCPU 数对齐），并加 --limit-concurrency 500 防止过载。另外把 pydantic 的 response_model 从完整 ORM 模型改成手写 dict 组装，省掉一次完整的模型校验序列化，序列化耗时从 35ms 降到 8ms。

五、效果数据

压测条件：wrk，200 并发，持续 120s，商品 ID 随机取自 10 万个热门商品集合（保证缓存有命中也有回源）。

最终 P95 从 1240ms 降到 85ms，QPS 提升约 11.7 倍。Redis 命中率稳定在 94% 左右，L1 命中率约 41%。DB 单请求 SQL 从 23 条降到 4 条，慢查询日志里超过 100ms 的 SQL 归零。

六、踩坑与总结

几个值得记下来的点：

1. 先 profiling 再优化。我最开始差点直接上缓存，幸亏先跑了 py-spy，否则缓存会掩盖 N+1 问题，一旦缓存失效就是雪崩。
2. selectinload 不是万能的。评论这种一对多且数据量大的关联，一定要单独查并 limit，预加载会把自己坑死。
3. 本地缓存必须有上限。dict 当缓存是新手陷阱，TTLCache 或 cachetools.LRUCache 是底线。
4. 序列化也是热点。pydantic 的 response_model 在大对象上开销不小，能用轻量组装就别全量校验。
5. worker 数不是越多越好。uvicorn workers 一般对齐 vCPU 数，配合 limit-concurrency 做背压，比盲目堆 worker 有效。

性能调优这件事，数据永远比直觉靠谱。火焰图、慢查询日志、压测数字三件套配齐，问题基本藏不住。希望这篇记录对正在被 API 延迟折磨的你有用，欢迎评论区交流。