一、问题背景
事情是这样的。我们有一个商品详情接口 GET /api/v1/products/{id}，早期只有几千 QPS 的零头，怎么写都无所谓。后来接入了推荐位和 App 首页，流量涨到日均 200 万请求，峰值 QPS 大概 600 左右。然后监控开始报警：P99 从 200ms 一路爬到 1.8s，P95 也有 900ms。
诡异的地方在于：CPU 利用率只有 30% 左右，内存也正常，机器完全没跑满。这说明瓶颈不在计算，而在等待——大概率是 IO。
线上环境大致是：
- Python 3.11.6
- FastAPI 0.109.2 + Uvicorn 0.27.1（4 worker，gunicorn 管理）
- SQLAlchemy 2.0.25（同步 Session，跑在 FastAPI 的线程池里）
- MySQL 8.0.35，主从架构，读走从库
- Redis 7.2.3
- 部署在 4 台 8C16G 的容器上
接口逻辑本身不复杂：查商品主表、查 SKU 列表、查库存、查促销标签、查用户是否收藏，拼装后返回。听起来就五个查询，但实际不是。
二、定位：先用 py-spy，再用 SQLAlchemy 事件
2.1 py-spy 抓火焰图
线上排查我第一反应是 py-spy，因为它可以 attach 到运行中的进程，几乎零开销：
# 在容器内 attach，采样 30 秒，生成火焰图
py-spy record -o /tmp/profile.svg \
--pid 12345 \
--duration 30 \
--rate 200 \
--threads
火焰图上非常明显：大量时间堆在 socket.recv 和 pymysql.connections.read 上，也就是在等数据库返回。--threads 参数很关键，因为我们的 SQLAlchemy 是同步的，跑在 anyio 的线程池里，不加这个参数看不到那些线程。
这一步就基本确认了方向：不是 Python 代码慢，是数据库交互慢。
2.2 SQLAlchemy 事件钩子统计慢查询
接下来要知道具体是哪些 SQL 慢、慢多少、执行了几次。我用 SQLAlchemy 的 before_cursor_execute / after_cursor_execute 事件做了一个轻量统计器，按 SQL 模板聚合：
# app/db_profiler.py
import time
import re
from collections import defaultdict
from sqlalchemy import event
from sqlalchemy.engine import Engine
_stats = defaultdict(lambda: {"count": 0, "total_ms": 0.0, "max_ms": 0.0})
# 把具体参数替换成 ?，方便按模板聚合
_PARAM_RE = re.compile(r"\b\d+\b|'[^']*'")
def _normalize(statement: str) -> str:
return _PARAM_RE.sub("?", statement)[:300]
@event.listens_for(Engine, "before_cursor_execute")
def _before(conn, cursor, statement, parameters, context, executemany):
context._query_start = time.perf_counter()
@event.listens_for(Engine, "after_cursor_execute")
def _after(conn, cursor, statement, parameters, context, executemany):
elapsed_ms = (time.perf_counter() - context._query_start) * 1000
key = _normalize(statement)
s = _stats[key]
s["count"] += 1
s["total_ms"] += elapsed_ms
s["max_ms"] = max(s["max_ms"], elapsed_ms)
def dump_top(n: int = 20):
rows = sorted(_stats.items(), key=lambda kv: kv[1]["total_ms"], reverse=True)[:n]
for sql, s in rows:
print(f"{s['count']:>6} 次 | 总 {s['total_ms']:.0f}ms | "
f"max {s['max_ms']:.0f}ms | {sql}")
注意：这个 profiler 只开在预发环境，线上全量打开事件钩子会有额外开销（实测约 3%~5%），别长期挂生产。
跑 1000 次请求后 dump_top() 的输出让我有点想骂人：
5000 次 | 总 41200ms | max 89ms | SELECT ... FROM product_sku WHERE product_id = ?
5000 次 | 总 18600ms | max 41ms | SELECT ... FROM inventory WHERE sku_id = ?
1000 次 | 总 9200ms | max 22ms | SELECT ... FROM product WHERE id = ?
1000 次请求，SKU 查询执行了 5000 次——平均每个商品 5 个 SKU，然后每个 SKU 又单独查一次库存。这就是经典的 N+1：先查 SKU 列表，再循环查库存。一个请求打出去 7~11 条 SQL，P99 不炸才怪。
三、方案设计
定位清楚后，方案其实就三件事，按收益排序：
- 消灭 N+1：SKU 和库存改成一次
IN查询批量取，5+5 条变 2 条。 - 加 Redis 缓存：商品详情是典型的读多写少，缓存整个响应体，TTL 60s + 主动失效。
- 补索引：
inventory.sku_id上居然没索引（历史遗留），顺手加上。
缓存策略上我选了 Cache-Aside，没有用 @lru_cache 之类的进程内缓存，因为 4 个 worker × 4 台机器，进程内缓存一致性太难管。key 设计为 product:detail:v2:{product_id}，带版本号前缀方便整体失效。TTL 60 秒是妥协：促销标签有时效性，太长会脏；太短缓存命中率上不去。另外商品变更时通过消息队列发失效事件，删 key 而不是更新 key（避免并发写覆盖）。
四、核心实现
4.1 批量查询消灭 N+1
改造前的伪代码长这样：
skus = session.query(SKU).filter(SKU.product_id == pid).all()
for sku in skus: # N+1 元凶
sku.inventory = session.query(Inventory).filter(
Inventory.sku_id == sku.id).first()
改成两次查询 + 内存组装：
# app/services/product.py
from sqlalchemy import select
from sqlalchemy.orm import selectinload
def get_product_detail(session, product_id: int):
stmt = (
select(Product)
.where(Product.id == product_id)
.options(
# 一次 JOIN 拿 SKU，避免 lazy load
selectinload(Product.skus).selectinload(SKU.inventory),
)
)
product = session.execute(stmt).scalar_one_or_none()
if product is None:
return None
# 促销标签批量取，一条 IN 查询
sku_ids = [s.id for s in product.skus]
tags = session.execute(
select(PromotionTag)
.where(PromotionTag.sku_id.in_(sku_ids))
).scalars().all()
tag_map = {}
for t in tags:
tag_map.setdefault(t.sku_id, []).append(t.name)
return {
"id": product.id,
"title": product.title,
"skus": [
{
"id": s.id,
"price": str(s.price),
"stock": s.inventory.available if s.inventory else 0,
"tags": tag_map.get(s.id, []),
}
for s in product.skus
],
}
selectinload 会先查主表，再用一条 WHERE product_id IN (...) 查关联，比 joinedload 更适合一对多（不会产生笛卡尔积膨胀）。改造后单请求 SQL 数从 7~11 条降到固定 3 条。
4.2 Redis 缓存层
# app/cache.py
import json
import redis
from app.config import settings
pool = redis.ConnectionPool.from_url(
settings.REDIS_URL,
max_connections=50,
socket_timeout=0.2, # 200ms 超时，宁可穿透也不能拖死接口
socket_connect_timeout=0.1,
health_check_interval=30,
)
r = redis.Redis(connection_pool=pool)
CACHE_TTL = 60
KEY_PREFIX = "product:detail:v2:"
def get_cached_detail(product_id: int):
try:
raw = r.get(f"{KEY_PREFIX}{product_id}")
return json.loads(raw) if raw else None
except redis.RedisError:
return None # 缓存故障降级为直接查库
def set_cached_detail(product_id: int, data: dict):
try:
r.setex(f"{KEY_PREFIX}{product_id}", CACHE_TTL,
json.dumps(data, ensure_ascii=False))
except redis.RedisError:
pass # 写缓存失败不影响主流程
def invalidate(product_id: int):
try:
r.delete(f"{KEY_PREFIX}{product_id}")
except redis.RedisError:
pass
接口层：
@app.get("/api/v1/products/{product_id}")
def product_detail(product_id: int, session: Session = Depends(get_session)):
cached = get_cached_detail(product_id)
if cached is not None:
return cached
data = get_product_detail(session, product_id)
if data is None:
raise HTTPException(status_code=404)
set_cached_detail(product_id, data)
return data
socket_timeout=0.2 是我特意压低的：缓存挂了也不能让接口跟着挂，200ms 拿不到就当没命中。这个降级逻辑后来真救过一次——Redis 主从切换那 40 秒，接口 P99 只涨了 60ms。
五、踩坑与二次优化
坑一：缓存击穿。 热门商品 key 过期瞬间，几十个并发同时打到数据库。压测时 P99 出现尖刺。加了个简单的互斥：命中失败时 SET NX 抢锁，抢到的查库回填，没抢到的 sleep 20ms 重试一次读缓存，还读不到就直接查库（保证可用性优先）。
坑二：selectinload 的 IN 参数膨胀。 有个类目下的商品 SKU 特别多（单商品 300+ SKU），IN (300 个 id) 的 SQL 在从库上执行计划走偏，反而慢了。后来对 SKU 数量加了上限保护，超过 200 的走分页接口，详情页只返回前 50 个。
坑三：JSON 序列化开销。 一开始用 json.dumps 序列化 Decimal 类型的价格直接报错，改成 str(s.price) 输出。另外大对象（约 12KB）序列化本身要 1~2ms，后来把响应体精简了一版，砍掉不用的字段，降到 4KB 左右。
坑四：索引加了但没生效。 inventory.sku_id 加索引后观察慢查询日志，发现还是有全表扫。原因是该字段类型是 varchar(32)，而代码里传的是 int，MySQL 隐式转换导致索引失效。把字段类型和代码统一成 bigint 后解决。这个坑花了我半天。
六、效果数据
压测用 wrk，固定 200 并发、持续 3 分钟，对比改造前后（数据为 3 轮中位数）：
| 指标 | 改造前 | 改造后 | 变化 |
|---|---|---|---|
| QPS | 320 | 1450 | +353% |
| P50 | 210ms | 42ms | -80% |
| P95 | 900ms | 120ms | -87% |
| P99 | 1800ms | 180ms | -90% |
| 单请求 SQL 数 | 7~11 | 1（缓存命中）/ 3（未命中） | - |
| 数据库 QPS | 约 2800 | 约 670 | -76% |
| 缓存命中率 | - | 93.4% | - |
线上灰度一周后，P99 稳定在 170~200ms 区间，CPU 从 30% 涨到 45%（因为 QPS 上去了），数据库从库负载明显下降，之前偶发的慢查询告警也消失了。
七、总结
回头看，这次调优没有用什么高深技巧，核心就是先测量再动手：py-spy 告诉我时间花在等 IO，SQLAlchemy 事件钩子告诉我具体是哪条 SQL 在重复执行。如果一开始就拍脑袋加缓存，N+1 的问题会被缓存掩盖，缓存一失效照样雪崩。
几条可以复用的经验：
- 同步 ORM + FastAPI 的组合下，
py-spy record --threads是定位 IO 等待最快的工具，没有之一。 - N+1 用
selectinload解决一对多，用批量IN解决跨表关联，别用joinedload处理一对多。 - 缓存一定要有超时和降级，
socket_timeout给到 200ms 以内，缓存不可用时接口要能裸奔。 - 加完索引记得验证执行计划，类型隐式转换是索引失效的常见原因。
如果你们接口也有「CPU 不高但延迟很高」的症状，大概率和我一样，先去数一数一次请求到底打了几条 SQL 吧。
