一、问题背景
事情是这样的。我们有个内部服务，主体是 FastAPI 写的查询接口，主要给前端做商品聚合展示：一个请求进来，要查商品基础信息、SKU 列表、库存、营销标签，拼成一个嵌套 JSON 返回。上线初期数据量小，没人管，后来商品表涨到 300 万行，接口就开始不对劲了。
监控上看到的数字：P95 响应时间 1.8s，P99 到了 3.2s。用 Locust 压测，50 并发的时候错误率就飙到 8%，全是超时（网关 2s 超时）。运维天天在群里 @ 我，前端说页面转圈转到怀疑人生。
这个接口逻辑本身不复杂，所以我很确定问题出在实现细节上，不是架构问题。于是开始定位。
二、环境与版本
先把环境交代清楚，不然数据没参考价值：
- Python 3.11.6
- FastAPI 0.109.2，uvicorn 0.27.1（4 worker，gunicorn 管理）
- Flask 侧还有一个老接口服务：Flask 2.3.3 + gunicorn 21.2.0
- PostgreSQL 14.10，psycopg2-binary 2.9.9
- Redis 7.2.4，redis-py 5.0.1
- SQLAlchemy 2.0.25（同步 ORM，没用 async，这是历史包袱）
- 压测机：Locust 2.20，8C16G，和服务同内网
- 服务机：4C8G 容器，4 个 uvicorn worker
三、定位：profiling 三板斧
3.1 第一板斧：cProfile 抓函数级耗时
先别猜，直接上 cProfile。我写了个脚本，直接调用接口对应的 handler 函数，跑 100 次取累计耗时：
import cProfile
import pstats
from app.api.product import get_product_detail
def run():
for _ in range(100):
get_product_detail(product_id=123456)
profiler = cProfile.Profile()
profiler.enable()
run()
profiler.disable()
stats = pstats.Stats(profiler)
stats.sort_stats("cumulative").print_stats(25)
输出里最扎眼的两行：
ncalls tottime cumtime function
100 0.021 142.3 get_product_detail
3100 0.890 118.7 sqlalchemy.orm.query.Query.all
100 0.012 21.4 jsonable_encoder
Query.all 被调了 3100 次——100 个请求调了 3100 次查询，平均每个请求 31 次 SQL。典型的 N+1。
3.2 第二板斧：py-spy 看线上真实火焰图
cProfile 有侵入性，线上不敢直接开。用 py-spy 采样：
py-spy record -o profile.svg --pid --duration 30
火焰图显示 60% 以上的采样落在 psycopg2 的 execute 等待上，剩下的一部分在 json.dumps。也就是说，时间基本都花在等数据库和序列化上，CPU 本身没怎么算。
3.3 第三板斧：SQL 日志确认 N+1 来源
打开 SQLAlchemy 的 echo，一个请求打出来 31 条 SQL，结构是这样的：
SELECT * FROM product WHERE id = 123456;
SELECT * FROM sku WHERE product_id = 123456; -- 1 次
SELECT * FROM inventory WHERE sku_id = 1; -- 每个 sku 一次
SELECT * FROM inventory WHERE sku_id = 2;
... -- 共 20 次
SELECT * FROM tag WHERE product_id = 123456;
SELECT * FROM tag_rel WHERE tag_id = ...; -- 又是 N 次
问题很清楚：ORM 里用了 lazy loading，product.skus、sku.inventory、product.tags 全是懒加载，循环里一访问就发一条 SQL。20 个 SKU 就是 20 次库存查询，加上标签关联，31 条 SQL 完全对得上。
四、方案设计与核心实现
定位清楚后，改动分三块：查询优化、缓存、序列化。
4.1 查询优化：干掉 N+1，加索引
第一步，把懒加载改成显式 join + selectinload。SQLAlchemy 2.0 里 selectinload 对一对多关系很合适，它会用一条 IN 查询批量加载，而不是逐条查：
from sqlalchemy.orm import selectinload
from sqlalchemy import select
def fetch_product_detail(db, product_id: int):
stmt = (
select(Product)
.options(
selectinload(Product.skus).selectinload(Sku.inventory),
selectinload(Product.tags),
)
.where(Product.id == product_id)
)
return db.execute(stmt).scalar_one_or_none()
改完后 SQL 数量从 31 条降到 4 条：product 一条、sku 一条、inventory 一条（IN 批量）、tag 一条。
第二步，检查执行计划。EXPLAIN ANALYZE 发现 inventory.sku_id 和 tag_rel.product_id 上没有索引，走的是 Seq Scan，单条就要 40ms+。补索引：
CREATE INDEX CONCURRENTLY idx_inventory_sku_id ON inventory (sku_id);
CREATE INDEX CONCURRENTLY idx_tag_rel_product_id ON tag_rel (product_id);
CREATE INDEX CONCURRENTLY idx_sku_product_id ON sku (product_id);
注意线上用 CONCURRENTLY，不然建索引会锁表。补完后这几条查询都走 Index Scan，单条从 40ms 降到 2ms 以内。
4.2 缓存策略：Redis + 分级 TTL
商品详情是典型的读多写少，非常适合缓存。但直接整对象缓存有个坑：库存变化频繁，缓存久了会超卖展示。我的策略是分级：
- 商品基础信息 + SKU + 标签：TTL 300s，变更时主动删
- 库存：TTL 10s，短缓存兜住瞬时流量
key 设计用 product:detail:{id}:v2，带版本号方便整体失效。代码：
import json
import redis
from app.core.config import REDIS_URL
r = redis.from_url(REDIS_URL, decode_responses=True)
def get_product_cached(db, product_id: int):
cache_key = f"product:detail:{product_id}:v2"
cached = r.get(cache_key)
if cached:
return json.loads(cached)
data = fetch_product_detail(db, product_id)
if data is None:
# 防穿透：空结果也缓存，TTL 短一点
r.setex(cache_key, 60, json.dumps(None))
return None
payload = serialize_product(data)
r.setex(cache_key, 300, json.dumps(payload, ensure_ascii=False))
return payload
更新商品时删 key，这里有个经典坑我后面会说。
4.3 序列化优化：去掉 jsonable_encoder
FastAPI 返回 Pydantic 模型时会走 jsonable_encoder，对嵌套结构递归转换，火焰图里那 20% 就是它。既然缓存里已经是 dict 了，直接返回 JSONResponse 或 ORJSONResponse 绕过这层：
from fastapi.responses import ORJSONResponse
@app.get("/api/product/{product_id}")
def product_detail(product_id: int, db: Session = Depends(get_db)):
data = get_product_cached(db, product_id)
if data is None:
raise HTTPException(status_code=404, detail="product not found")
return ORJSONResponse(data)
ORJSONResponse 底层是 orjson，序列化速度比标准库 json 快 3~5 倍，配合 pip install orjson 即可。
4.4 连接池调参
顺手把 SQLAlchemy 连接池也调了。默认 pool_size=5，4 个 worker 就是 20 个连接，压测时 pool_timeout 报错。改成：
engine = create_engine(
DATABASE_URL,
pool_size=10,
max_overflow=20,
pool_pre_ping=True,
pool_recycle=1800,
)
PostgreSQL 的 max_connections 是 200，4 worker × (10+20) = 120，留了余量。
五、踩坑与优化
坑一：缓存删除的时序问题。 最开始是先更新 DB 再删缓存，压测时发现偶尔读到旧数据。原因是并发下"读请求回填缓存"和"写请求删缓存"交错。后来改成延迟双删：更新后删一次，500ms 后再删一次，脏数据窗口基本消除。更严谨的做法是订阅 binlog，但当前量级没必要。
坑二：selectinload 不是万能的。 一开始我对 tag 也用了 selectinload，但 tag 关联表数据量巨大，IN 列表过长导致查询变慢。后来改成 join + 只取需要的字段，反而更快。ORM 的优化要看着数据分布来，别一把梭。
坑三：Flask 老服务的同类问题。 我们还有个 Flask 2.3.3 的接口服务，同样有 N+1。Flask-SQLAlchemy 里对应的是 lazy="selectin" 配置：
class Product(db.Model):
__tablename__ = "product"
skus = db.relationship("Sku", lazy="selectin", backref="product")
另外 Flask 侧用 flask-caching 接 Redis，装饰器一行搞定：
from flask_caching import Cache
cache = Cache(app, config={"CACHE_TYPE": "RedisCache", "CACHE_REDIS_URL": REDIS_URL})
@app.route("/api/product/")
@cache.cached(timeout=300, key_prefix=lambda: f"product:detail:{request.view_args['pid']}")
def product_detail(pid):
...
改完 Flask 那个接口 P95 从 900ms 降到 180ms，效果同样明显。
坑四：压测机自己成了瓶颈。 第一次压测 QPS 上不去，查了半天发现 Locust 单进程 CPU 打满了。后来用 --processes 4 多进程压测，数据才可信。压测工具的瓶颈经常被忽略。
六、效果数据
压测条件：同一台 Locust 机，4 进程，持续 5 分钟，目标接口 /api/product/{id}，商品 ID 随机取自 300 万行。
| 指标 | 优化前 | 优化后 |
|---|---|---|
| P50 | 620ms | 45ms |
| P95 | 1800ms | 120ms |
| P99 | 3200ms | 210ms |
| QPS（错误率<1%） | 50 | 420 |
| 单请求 SQL 数 | 31 | 4（命中缓存时 0） |
| 缓存命中率 | - | 93.6% |
线上灰度后观察一周，P95 稳定在 110~130ms，数据库 CPU 从 75% 降到 22%，效果符合预期。
七、总结
这次调优没什么高深技巧，核心就三件事：先用 profiling 工具把问题钉死在具体函数上，别猜；N+1 是 ORM 项目最常见的性能杀手，selectinload 加索引基本能解决八成；缓存要按数据变更频率分级，别一个 TTL 打天下。
几个可以复用的经验：
- cProfile 适合离线复现，py-spy 适合线上采样，两个配合用。
EXPLAIN ANALYZE一定要看，索引缺失在数据量上来后是断崖式的影响。- 缓存 key 带版本号，出问题能整体切换，比逐个删靠谱。
- 压测先确认压测工具本身没到瓶颈，不然优化方向会跑偏。
如果你们项目也是 FastAPI/Flask + 同步 ORM 的组合，建议先跑一遍 profiling，大概率能挖出几个 N+1。异步 ORM 能进一步压延迟，但那是另一个话题了，这次先不动。
