一、问题背景：一个"看起来很健康"的慢接口
事情起因很简单。我们有个订单查询接口 /api/v1/orders/summary，前端每次进首页都要调它，返回用户最近 30 天的订单聚合数据（订单数、总金额、最近 5 单）。上线初期 QPS 不到 50，没人关注。直到做活动，QPS 冲到 200 左右，监控开始报警：
-
P99 延迟：1.8s（SLO 要求 = datetime.now() - timedelta(days=30)
).all()result = []
for order in orders: # 每次循环都查一次 items
items = db.query(OrderItem).filter(
OrderItem.order_id == order.id
).all()
result.append({"order": order, "items": items})
return result
30 天有 40 个订单，就是 1 + 40 = 41 次查询。改成一次性 JOIN + 预加载：
```python
# 优化后：使用 selectinload 批量加载
from sqlalchemy.orm import selectinload
def get_order_summary(user_id: int):
thirty_days_ago = datetime.now() - timedelta(days=30)
orders = (
db.query(Order)
.options(selectinload(Order.items)) # 一次 IN 查询搞定所有 items
.filter(
Order.user_id == user_id,
Order.created_at >= thirty_days_ago
)
.order_by(Order.created_at.desc())
.limit(100)
.all()
)
return orders
selectinload 会生成 WHERE order_id IN (...) 的批量查询，从 41 次降到 2 次。同时补上复合索引：
CREATE INDEX CONCURRENTLY idx_orders_user_created
ON orders (user_id, created_at DESC);
CONCURRENTLY 是关键，生产环境建索引不锁表。
4.2 缓存：Redis 三级策略
聚合数据不需要实时，5 分钟缓存完全可接受。用 Redis 做缓存，key 设计成 order:summary:{user_id}:{date}，同时加随机 TTL 防雪崩：
import json
import random
import redis
from functools import wraps
r = redis.Redis(host="localhost", port=6379, db=0,
socket_timeout=0.1, decode_responses=True)
def cache_result(prefix: str, ttl: int = 300):
def decorator(func):
@wraps(func)
def wrapper(user_id: int, *args, **kwargs):
key = f"{prefix}:{user_id}"
try:
cached = r.get(key)
if cached:
return json.loads(cached)
except redis.RedisError:
pass # 缓存挂了降级走 DB，别让缓存拖垮服务
result = func(user_id, *args, **kwargs)
try:
# TTL 加随机抖动，避免同一时刻集体失效
jitter = random.randint(0, 60)
r.setex(key, ttl + jitter, json.dumps(result, default=str))
except redis.RedisError:
pass
return result
return wrapper
return decorator
@cache_result("order:summary", ttl=300)
def get_order_summary(user_id: int):
...
4.3 消除同步阻塞
profiling 还暴露了一个隐藏问题：代码里有个同步的 HTTP 调用（查用户等级），在 async 的 FastAPI 里直接 requests.get，会把事件循环堵死。改成 httpx.AsyncClient：
import httpx
async def get_user_level(user_id: int) -> str:
async with httpx.AsyncClient(timeout=0.5) as client:
resp = await client.get(f"http://user-svc/level/{user_id}")
return resp.json()["level"]
五、踩坑与优化
坑 1：selectinload 不是万能的。 一开始我对一个 10 万行的大表用了 selectinload，结果 IN 列表太长，PG 直接报错。后来改成 JOIN + 分页，或者限制 limit。
坑 2：py-spy 在容器里要加 --nonblocking。 否则采样会卡住进程。另外容器需要 SYS_PTRACE 权限：
# docker-compose.yml
services:
api:
cap_add:
- SYS_PTRACE
坑 3：Flask 那边的线程池打满。 Flask 用 gthread，workers=4、threads=8，最大并发 32。压测到 QPS 300 就开始排队。后来把数据库连接池调大（pool_size=20, max_overflow=40），并把 pool_pre_ping=True 打开，避免连接失效。
坑 4：缓存穿透。 用户传了不存在的 user_id，每次都打到 DB。加了空值缓存（TTL 60s）+ 布隆过滤器。
六、效果数据
用 wrk 压测，脚本如下：
# 压测脚本
wrk -t8 -c200 -d60s --latency \
-s post.lua \
http://localhost:8000/api/v1/orders/summary
post.lua 里随机生成 user_id 避免全命中缓存：
wrk.method = "POST"
wrk.headers["Content-Type"] = "application/json"
request = function()
local uid = math.random(1, 10000)
return wrk.format(nil, nil, nil, string.format('{"user_id": %d}', uid))
end
优化前后对比（QPS 200 场景）：
| 指标 | 优化前 | 优化后 | 提升 |
|---|---|---|---|
| P50 | 380ms | 45ms | 88%↓ |
| P99 | 1820ms | 120ms | 93%↓ |
| QPS | 210 | 1620 | 7.7x |
| DB QPS | 4200 | 540 | 87%↓ |
| 缓存命中率 | - | 94.3% | - |
FastAPI 和 Flask 两套服务的优化后数据：
| 框架 | P99 | QPS | 备注 |
|---|---|---|---|
| FastAPI (uvicorn 4w) | 120ms | 1620 | async 版本 |
| Flask (gunicorn 4w×8t) | 145ms | 1180 | 同步版本，受 GIL 限制 |
Flask 因为同步模型，QPS 上限就是比 FastAPI 低一截，这是架构决定的，不是代码问题。
七、总结
这次调优最大的收获是：别急着优化，先让数据说话。我一开始以为是 Python 慢，差点去上 Cython，结果 profiling 一跑，全是 IO 等待。
几条可复用的经验：
- py-spy + cProfile 是标配，生产用 py-spy，本地复现用 cProfile。
- N+1 是 API 性能的头号杀手，SQLAlchemy 用
selectinload或joinedload。 - 缓存要加降级、抖动、空值保护，否则缓存本身会成为故障点。
- FastAPI 里别写同步阻塞代码，一个
requests.get能堵死整个事件循环。 - 压测要模拟真实流量，全命中缓存的压测数据毫无意义。
最后提醒一句：优化完记得回归测试，我改完 selectinload 后有个订单状态字段的排序变了，差点上线出事故。性能优化，功能正确永远是第一位的。
