FastAPI+Flask接口从1.2s压到80ms：一次真实的性能瓶颈定位与优化

一、问题背景

上个月接手了一个电商中台项目，技术栈有点混搭：对外的BFF层用 FastAPI 0.110.0 + Uvicorn 0.29.0，内部几个老服务还是 Flask 3.0.2 + Gunicorn 21.2.0。平时量不大，直到某次活动预热，监控开始疯狂告警：

• /api/v1/product/{id} 接口 P99 从 200ms 涨到 1.2s
• QPS 到 60 左右就开始雪崩，Uvicorn worker 全部打满
• CPU 使用率却只有 40% 左右，明显是在等 IO

这是个典型的“不是算不过来，是等不过来”的场景。接口逻辑很简单：查商品基本信息、查SKU列表、查库存、查促销标签，拼装后返回。问题就出在这几个“查”上。

二、环境与版本

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

压测工具用 wrk 4.2.0，命令固定为：

wrk -t8 -c200 -d60s --latency http://127.0.0.1:8000/api/v1/product/10086

三、定位：先用 py-spy 把火焰图打出来

优化第一步永远是先测量，别猜。我见过太多人上来就加缓存，结果缓存了本来就不慢的东西。

对 FastAPI 进程用 py-spy 采样（生产环境可以直接用，开销极低）：

# 找到 uvicorn worker pid
ps -ef | grep uvicorn

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

火焰图一出来，问题清清楚楚：超过 70% 的采样点卡在 psycopg2 的 execute 上，而且调用栈里反复出现 product_detail → get_skus → get_stock → get_promotions 这几个函数，每个都是独立的 session.execute。

典型 N+1：主查询拿到商品后，SKU、库存、促销各发一条 SQL，一个请求打了 4 条 SQL，其中 SKU 和库存还按 SKU 数量循环查。一个商品 30 个 SKU，就是 60+ 条 SQL。P99 1.2s 一点都不冤。

顺手也测了下 Flask 那个老服务，用 flask-profiler 或者更简单的 cProfile：

python -m cProfile -o flask.prof app.py
# 用 snakeviz 看
snakeviz flask.prof

Flask 侧问题不一样，是 requests 同步调用下游服务 + 数据库连接池只有默认的 5，高并发下大量时间耗在等连接。

四、方案设计

定位清楚后，优化分三层：

1. 数据库层：消灭 N+1，用 selectinload 预加载关联，合并查询
2. 缓存层：热点商品详情用 Redis 缓存，设置合理的 TTL 和空值缓存
3. 服务层：FastAPI 侧保持 async，但把 ORM 换成异步会话；Flask 侧调大连接池

五、核心实现

5.1 消灭 N+1：SQLAlchemy 2.0 的 selectinload

优化前（伪代码，就是问题的根源）：

# 每条 SQL 都单独发，N+1 的经典写法
product = session.get(Product, pid)
skus = session.query(Sku).filter(Sku.product_id == pid).all()
for sku in skus:
    sku.stock = session.query(Stock).filter(Stock.sku_id == sku.id).first()
promotions = session.query(Promotion).filter(Promotion.product_id == pid).all()

优化后，用 SQLAlchemy 2.0 的 selectinload 一次性预加载：

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

async def get_product_detail(session: AsyncSession, pid: int) -> dict:
    stmt = (
        select(Product)
        .where(Product.id == pid)
        .options(
            # selectinload 会额外发一条 IN 查询，而不是 N 条
            selectinload(Product.skus).selectinload(Sku.stock),
            selectinload(Product.promotions),
        )
    )
    result = await session.execute(stmt)
    product = result.scalar_one_or_none()
    if product is None:
        return None
    return serialize(product)

这样 SQL 从 60+ 条降到 4 条（主查询 + skus IN 查询 + stock IN 查询 + promotions IN 查询），而且全是走主键/索引的批量查询。

踩坑：selectinload 和 joinedload 别乱混用。joinedload 在集合关联上会产生笛卡尔积，SKU 30 个 × 促销 5 个 = 150 行结果集，反而更慢。一对多用 selectinload，多对一用 joinedload，这是铁律。

5.2 缓存策略：Redis + 空值 + 随机 TTL

数据库优化后 P99 从 1.2s 降到 ~180ms，但热点商品还是每次都打 DB。加缓存：

import json
import random
import redis.asyncio as aioredis

redis = aioredis.from_url(
    "redis://127.0.0.1:6379/0",
    max_connections=50,
    socket_timeout=0.5,
    decode_responses=True,
)

CACHE_KEY = "product:detail:{pid}"
NULL_FLAG = "__NULL__"

async def get_product_cached(session, pid: int):
    key = CACHE_KEY.format(pid=pid)
    cached = await redis.get(key)
    if cached is not None:
        # 空值缓存，防穿透
        if cached == NULL_FLAG:
            return None
        return json.loads(cached)

    data = await get_product_detail(session, pid)
    if data is None:
        # 空值缓存 60s，避免恶意 id 打穿 DB
        await redis.set(key, NULL_FLAG, ex=60)
        return None

    # TTL 加随机抖动，防止缓存雪崩同时失效
    ttl = 300 + random.randint(0, 60)
    await redis.set(key, json.dumps(data, ensure_ascii=False), ex=ttl)
    return data

几个关键点：

• 空值缓存：防缓存穿透，恶意请求不存在的 id 直接命中 NULL_FLAG
• TTL 抖动：300~360s 随机，防雪崩
• socket_timeout=0.5：Redis 挂了不能拖垮接口，超时就降级走 DB
• 序列化用 ensure_ascii=False：中文商品名不转义，体积小 30%

缓存失效走主动删除：商品更新时 await redis.delete(key)，配合 TTL 兜底。

5.3 Flask 侧：连接池 + 超时

Flask 老服务没法大改，做两件小事：

# SQLAlchemy 连接池调优
engine = create_engine(
    DATABASE_URL,
    pool_size=20,          # 默认 5，太小
    max_overflow=40,       # 峰值可临时扩到 60
    pool_pre_ping=True,    # 自动剔除失效连接
    pool_recycle=1800,     # 30 分钟回收，防 PG 端断连
    pool_timeout=3,        # 拿不到连接 3s 就报错，别无限等
)

# 下游请求加超时 + 重试
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

session = requests.Session()
retry = Retry(total=2, backoff_factor=0.1, status_forcelist=[502, 503, 504])
session.mount("http://", HTTPAdapter(max_retries=retry, pool_connections=50, pool_maxsize=50))

resp = session.get(url, timeout=(0.3, 1.0))  # 连接0.3s，读1.0s

pool_timeout=3 这个参数特别重要。默认是 30s，高并发下线程全卡在等连接上，接口直接雪崩。设成 3s 快速失败，配合上游熔断反而稳。

六、踩坑与优化

几个真实踩过的坑，比代码本身更值钱：

1. selectinload 的 IN 参数超过 32767 会报错。PostgreSQL 对 IN 数量有限制，商品 SKU 特别多时要分批。我们后来对超过 500 个 SKU 的商品做了分页返回。

2. Uvicorn workers 不是越多越好。一开始设了 8 个 worker，结果 4 核机器上下文切换严重，QPS 反而下降。最终稳定在 workers=4（等于 CPU 核数），配合 async 已经能吃满。

3. Redis 缓存和 DB 数据不一致。有次商品改价，缓存没删干净，用户看到旧价格。后来改成先更新 DB，再删缓存（不是更新缓存），并且删除失败要记日志告警。

4. 压测环境和生产差距大。本地压测 QPS 1500，上线只有 800。原因是本地 Redis 和 PG 都在同机，网络延迟为 0。压测一定要贴近真实网络拓扑。

5. json.dumps 是 CPU 瓶颈。缓存命中后序列化大对象反而慢，后来对超大商品详情用 orjson 替换，序列化耗时降了 60%。

七、效果数据

优化前后用同样的 wrk -t8 -c200 -d60s 压测，结果对比：

注意 CPU 从 40% 涨到 68%，这不是变差了，而是之前 CPU 在空转等 IO，现在真正在干活。QPS 提升 18 倍的同时 CPU 还没打满，说明还有余量。

Flask 老服务单独压测，QPS 从 210 提到 560，P99 从 900ms 降到 210ms，主要靠连接池和超时快速失败。

八、总结

这次优化的核心就三步：py-spy 定位 → selectinload 消灭 N+1 → Redis 缓存兜底。如果只能记一句话，那就是：先 profile，再动手。火焰图上那 70% 的 psycopg2 采样点，比任何猜测都可靠。

几个可以直接抄走的经验：

• 一对多关联用 selectinload，别用 joinedload
• 缓存必须配空值缓存 + TTL 抖动 + 超时降级
• 连接池 pool_timeout 一定要调小，快速失败好过慢慢雪崩
• Uvicorn workers 设成 CPU 核数即可，别贪多
• 压测环境要贴近生产，本地数据会骗人

代码不复杂，难的是知道该改哪里。希望这篇能帮你少走点弯路。