一、问题背景
事情是这样的。我们有个内部商品中台，对外提供 /api/v1/products/search 接口，Flask 老服务，后来新业务线用 FastAPI 重写了一版网关层，两套栈并存。上线三个月后，运营那边开始抱怨「商品列表页打开要转圈好几秒」。
我先看了一眼监控：接口 P99 延迟 2.3s，平均 860ms，QPS 峰值大概 120。数据库那边更惨，PostgreSQL 的 CPU 使用率长期在 85%~92% 之间晃，慢查询日志一天刷出几万条。这已经不是「优化一下」的问题了，是快要出事故了。
目标定得很明确：P99 压到 300ms 以内，QPS 撑到 1000 以上，数据库 CPU 降到 40% 以下。
二、环境与版本
先把家底交代清楚，不然复现不了：
- Python 3.11.6
- FastAPI 0.109.2 + Uvicorn 0.27.1（网关层）
- Flask 2.3.3 + Gunicorn 21.2.0，worker 数 4，sync worker（老服务层）
- SQLAlchemy 2.0.25（ORM），psycopg2-binary 2.9.9
- PostgreSQL 14.10，16C32G，shared_buffers 8GB
- Redis 7.2.4，单机
- 压测工具：Locust 2.20.1
- Profiling：cProfile（内置）、py-spy 0.3.14、pg_stat_statements
三、方案设计
我的思路是三步走，先测量再动手，别凭感觉瞎改：
- 定位：用 py-spy 做在线采样，拿到火焰图；用 pg_stat_statements 抓最耗时的 SQL；用 cProfile 在压测环境复现。
- 分层优化：应用层解决 N+1 和同步阻塞，数据库层重建索引、改写慢 SQL，缓存层加 Redis 两级缓存。
- 验证：每改一项，用 Locust 跑同一套压测脚本，记录 P50/P95/P99 和 QPS，做前后对比。
这里有个原则我想强调一下：一次只改一个变量。我第一次图省事，索引和缓存一起上，结果 QPS 涨了但不知道是谁的功劳，后面回滚排查特别痛苦。
四、核心实现
4.1 Profiling：先看清楚谁在拖后腿
py-spy 是这次的头号功臣，因为它能 attach 到生产进程上采样，不用改代码不用重启：
# 采样 30 秒，生成火焰图
py-spy record -o profile.svg --pid $(pgrep -f "uvicorn app.main:app") --duration 30 --rate 200
# 直接看实时热点，top 模式
py-spy top --pid $(pgrep -f "gunicorn")
火焰图一出来，问题一目了然：sqlalchemy/orm/loading.py 占了 60% 以上的采样，而且集中在 _load_for_path 上。这就是典型的 N+1——查商品列表拿到 20 个商品，然后每个商品再单独查一次分类和库存。
再看 pg_stat_statements：
SELECT query, calls, mean_exec_time, total_exec_time
FROM pg_stat_statements
ORDER BY total_exec_time DESC
LIMIT 10;
排名第一的是一条 SELECT ... FROM product_sku WHERE product_id = $1，调用次数 40 万+，平均 3.2ms。单条不算慢，但乘以次数就是灾难。
4.2 数据库查询优化：干掉 N+1 + 重建索引
原来 Flask 里那段代码大概是这样，很经典的写法：
# 改造前：N+1 重灾区
products = Product.query.filter(Product.status == 1).limit(20).all()
result = []
for p in products:
skus = ProductSku.query.filter_by(product_id=p.id).all() # 每个商品一次查询
result.append({
"id": p.id,
"name": p.name,
"skus": [{"id": s.id, "price": s.price} for s in skus]
})
20 个商品 = 1 + 20 次查询，接口里还有分类、品牌，实际是 1 + 60 次。改成 SQLAlchemy 2.0 的 selectinload 预加载，一次 IN 查询搞定：
from sqlalchemy import select
from sqlalchemy.orm import selectinload
from sqlalchemy.ext.asyncio import AsyncSession
async def search_products(db: AsyncSession, keyword: str, limit: int = 20):
stmt = (
select(Product)
.where(Product.status == 1, Product.name.ilike(f"%{keyword}%"))
.options(
selectinload(Product.skus),
selectinload(Product.category),
)
.order_by(Product.sort_weight.desc())
.limit(limit)
)
result = await db.execute(stmt)
return result.scalars().unique().all()
查询次数从 61 次降到 3 次（商品、SKU、分类各一次 IN 查询）。
索引这块，ilike '%keyword%' 前置通配符走不了 B-Tree，我们加了 pg_trgm 扩展建 GIN 索引：
CREATE EXTENSION IF NOT EXISTS pg_trgm;
CREATE INDEX CONCURRENTLY idx_product_name_trgm
ON product (name gin_trgm_ops);
-- 复合索引，覆盖过滤+排序
CREATE INDEX CONCURRENTLY idx_product_status_weight
ON product (status, sort_weight DESC) INCLUDE (name, category_id);
注意 CONCURRENTLY，生产环境不加这个锁表能锁到你怀疑人生，我第一次就差点把线上搞挂。
4.3 缓存策略：Redis 两级缓存
不是所有数据都值得缓存。我们的策略是：
- 商品详情：热点数据，TTL 300s，主动失效
- 搜索结果：按 keyword + 分页做 key，TTL 60s（避免脏数据太久）
- 分类/品牌字典：几乎不变，TTL 3600s
用 cache-aside 模式，读的时候先查 Redis，miss 了查库再回填：
import json
import hashlib
from redis.asyncio import Redis
redis: Redis = None # 启动时初始化
async def get_product_detail(db: AsyncSession, product_id: int) -> dict:
cache_key = f"product:detail:{product_id}"
cached = await redis.get(cache_key)
if cached:
return json.loads(cached)
product = await db.get(Product, product_id, options=[selectinload(Product.skus)])
if not product:
return None
data = serialize_product(product)
await redis.setex(cache_key, 300, json.dumps(data, ensure_ascii=False))
return data
写操作时删缓存而不是更新缓存（避免并发写覆盖），并且加了个简单的防击穿：缓存 miss 时用 SET NX 抢一个短锁，抢不到就等 50ms 重试，最多 3 次。
4.4 同步阻塞：Flask 那层的坑
FastAPI 层是 async 的，但 Flask 层 Gunicorn 用的是 sync worker，4 个 worker 意味着并发上限就 4，一旦有慢请求全堵死。我们先把 Flask 侧的慢查询按上面同样方式改掉，然后把 Gunicorn 的 worker 换成 gthread，--threads 8，同时把 --timeout 从 30 调到 60：
gunicorn -w 4 -k gthread --threads 8 --timeout 60 \
--max-requests 2000 --max-requests-jitter 200 \
-b 0.0.0.0:8000 app:application
--max-requests 那个是为了防内存泄漏，跑久了 Python 进程 RSS 会涨，定期回收。
五、踩坑与优化
说几个真踩过的坑，都是血泪：
坑一：selectinload 和 limit 一起用。selectinload 是二次查询，如果你在子查询里也加了 limit，行为会很怪。我们一开始对 skus 加了 limit(5)，结果 SQLAlchemy 直接把 limit 应用到了整体，返回数据错乱。正确做法是用 with_loader_criteria 或者干脆在应用层截断。
坑二：缓存序列化开销。一开始用 json.dumps，商品对象有几十个字段，序列化耗时占了接口时间的 15%。后来换 orjson，同样的数据快了约 3 倍，pip install orjson 就完事。
坑三：Redis 连接池没配。默认连接池大小不够，压测到 800 QPS 时开始报 ConnectionError: Too many connections。改成 max_connections=200，socket_timeout=1，问题消失。
坑四：索引建了但没用上。GIN trgm 索引建完，EXPLAIN ANALYZE 一看还是 Seq Scan。原因是 ilike 的 pattern 参数是运行时拼的，planner 估不准。最后在查询里显式 SET LOCAL enable_seqscan = off 做了对比验证，确认索引生效后就没再动，实际是统计信息过期，ANALYZE product 之后 planner 就正常走了。
六、效果数据
Locust 压测脚本：100 并发用户，spawn rate 10/s，持续 5 分钟，请求体就是随机 keyword 搜商品列表。
| 指标 | 优化前 | 优化后 |
|---|---|---|
| P50 | 780ms | 62ms |
| P95 | 1.9s | 140ms |
| P99 | 2.3s | 180ms |
| QPS | 120 | 1100 |
| 错误率 | 0.3% | 0% |
| DB CPU | 90% | 25% |
| 单请求 SQL 数 | 61 | 3 |
分项贡献大概是这样：N+1 修复贡献了约 55% 的延迟下降，索引贡献 20%，缓存贡献 20%，Gunicorn 线程模型调整贡献 5%。缓存命中率稳定在 87% 左右，符合预期。
七、总结
这次调优最大的体会是：别猜，去测。我一开始凭直觉以为是网络或者框架开销，差点去折腾 uvicorn 的 worker 数，结果 py-spy 一采样，问题明明白白写在 ORM 里。
几个可以复用的经验：
- py-spy 在线采样对生产零侵入，是定位 Python 服务性能问题的第一选择。
- N+1 是 ORM 项目最常见的性能杀手，
selectinload/joinedload要成为肌肉记忆。 - 索引不是建了就行，
EXPLAIN ANALYZE验证 +ANALYZE更新统计信息缺一不可。 - 缓存要设计失效策略和防击穿，不然就是给自己埋雷。
- 压测一定要用同一套脚本、同样的并发模型，否则数据没有可比性。
目前这套方案跑了两个月，P99 稳定在 200ms 以内。下一步打算把搜索结果缓存改成布隆过滤器预判 + 本地 LRU 两级，进一步压 Redis 的 RT。有进展再写一篇。
