一、问题背景：一个"看起来没问题"的接口
去年接手了一个订单中台项目，技术栈有点混搭：主服务是 FastAPI 0.104 + SQLAlchemy 2.0，另有一个历史遗留的 Flask 2.3 报表服务共用同一套 MySQL 8.0 库。
上线三个月后监控开始报警：
/api/v1/orders接口 P99 从 220ms 涨到 1.2s- MySQL 主库 CPU 长期 80%~95%
- 高峰期 QPS 只有 320 左右就开始出现 502
接口逻辑本身很简单：根据用户ID查订单列表，附带商品信息和物流状态。代码只有几十行，看起来"没问题"。但恰恰是这种"看起来没问题"的接口，最容易藏着性能坑。
下面是我完整的排查和优化过程，环境版本先列清楚，方便你对照复现。
二、环境与版本
| 组件 | 版本 |
|---|---|
| Python | 3.11.6 |
| FastAPI | 0.104.1 |
| Flask | 2.3.3 |
| SQLAlchemy | 2.0.23 |
| Uvicorn | 0.24.0 |
| MySQL | 8.0.35 (InnoDB) |
| Redis | 7.2.3 |
| py-spy | 0.3.14 |
| wrk | 4.2.0 |
部署：4C8G 容器 × 3，Uvicorn worker 数 4。压测机同内网，避免网络干扰。
三、方案设计：先定位，再动手
我的原则是先量化，再优化，绝不凭感觉改代码。整体分三步：
- Profiling 定位：用 py-spy 采样火焰图 + cProfile 精确到函数
- 数据库层：找出慢查询，解决 N+1，补索引
- 缓存层：热点数据上 Redis，做多级缓存
3.1 用 py-spy 抓火焰图
线上服务不能随便重启，py-spy 的最大好处是无需侵入、无需重启：
# 找到 uvicorn 主进程 PID
ps -ef | grep uvicorn
# 采样 30 秒，生成火焰图
py-spy record -o profile.svg --pid 12345 --duration 30 --rate 200
火焰图一出来，问题一目了然：超过 60% 的采样堆栈卡在 sqlalchemy/orm/loading.py，也就是 ORM 加载阶段。再看下面的调用，全是 SELECT ... WHERE id = ? 的循环。
3.2 cProfile 精确定位
火焰图看趋势，cProfile 看细节。本地复现后跑一遍：
import cProfile
import pstats
from app.main import get_orders # 你的接口函数
profiler = cProfile.Profile()
profiler.enable()
for _ in range(50):
get_orders(user_id=10086)
profiler.disable()
stats = pstats.Stats(profiler).sort_stats("cumulative")
stats.print_stats(20)
输出里 session.execute 被调用 1 + N 次（N 是订单数），单次请求平均 47 次 SQL——典型的 N+1 查询。
四、核心实现
4.1 优化一：干掉 N+1 查询
原始代码长这样（简化版）：
# ❌ 优化前：N+1 查询
@app.get("/api/v1/orders")
async def get_orders(user_id: int, db: Session = Depends(get_db)):
orders = db.query(Order).filter(Order.user_id == user_id).all()
result = []
for o in orders:
# 每次循环都查一次商品
product = db.query(Product).filter(Product.id == o.product_id).first()
# 又查一次物流
logistics = db.query(Logistics).filter(Logistics.order_id == o.id).first()
result.append({
"order_no": o.order_no,
"product": product.name if product else None,
"status": logistics.status if logistics else None,
})
return result
40 个订单 → 1 + 40 + 40 = 81 次 SQL。
改成 selectinload 预加载 + 一次 join 拿物流：
# ✅ 优化后：预加载，2 次 SQL
from sqlalchemy.orm import selectinload
from sqlalchemy import select
@app.get("/api/v1/orders")
async def get_orders(user_id: int, db: AsyncSession = Depends(get_async_db)):
stmt = (
select(Order)
.options(selectinload(Order.product))
.options(selectinload(Order.logistics))
.where(Order.user_id == user_id)
.order_by(Order.created_at.desc())
.limit(50)
)
result = await db.execute(stmt)
orders = result.scalars().all()
return [
{
"order_no": o.order_no,
"product": o.product.name if o.product else None,
"status": o.logistics.status if o.logistics else None,
}
for o in orders
]
SQL 从 81 次降到 3 次（1 主 + 2 预加载），单接口数据库耗时从 680ms → 95ms。
4.2 优化二：补索引 + 覆盖索引
用 SHOW INDEX FROM orders 一看，user_id 上竟然没索引——早期数据量小没人发现，现在表已经 800 万行，全表扫描。
-- 组合索引，覆盖 WHERE + ORDER BY
ALTER TABLE orders ADD INDEX idx_user_created (user_id, created_at DESC);
-- 物流表按 order_id 查，补上
ALTER TABLE logistics ADD INDEX idx_order_id (order_id);
加了 idx_user_created 后，EXPLAIN 从 type=ALL 变成 type=ref，扫描行数从 320 万 → 42 行，这条 SQL 从 410ms → 3ms。
4.3 优化三：Redis 多级缓存
订单列表是典型读多写少场景，非常适合缓存。我用了"本地缓存 + Redis"两级：
import json
import redis.asyncio as aioredis
from cachetools import TTLCache
# L1：进程内缓存，5秒，抗瞬时热点
local_cache = TTLCache(maxsize=1000, ttl=5)
# L2：Redis，60秒
rds = aioredis.from_url("redis://localhost:6379/0", decode_responses=True)
async def get_orders_cached(user_id: int, db: AsyncSession):
key = f"orders:user:{user_id}"
# L1 命中
if key in local_cache:
return local_cache[key]
# L2 命中
cached = await rds.get(key)
if cached:
data = json.loads(cached)
local_cache[key] = data
return data
# 回源数据库
data = await query_orders_from_db(user_id, db)
await rds.setex(key, 60, json.dumps(data))
local_cache[key] = data
return data
关键细节：
- 缓存穿透：不存在的 user_id 缓存空列表
[]，TTL 设 30 秒 - 缓存雪崩：TTL 加随机抖动
60 + random.randint(0, 10) - 失效策略：订单状态变更时，主动
DEL对应 key
五、踩坑与优化：几个让我加班到凌晨的点
坑1：selectinload 和 limit 一起用会多查数据
selectinload 是分两次查询，主查询 limit 50 没问题，但子查询会把 50 条订单对应的所有 product 都捞出来。如果 product 表很大，反而变慢。解决：product 单独走缓存。
坑2：Flask 报表服务拖垮主库
那个 Flask 服务有个导出接口，一次 SELECT * FROM orders WHERE created_at > '2023-01-01' 拉 200 万行。它和 FastAPI 共用一个库，直接把主库 IO 打满。后来把它迁到只读从库，并加了 LIMIT 强制约束。
坑3：async 里混用同步 SQLAlchemy 会阻塞事件循环
一开始我把 Session 直接塞进 async def，结果并发上不去。必须用 AsyncSession + asyncmy 驱动，否则 py-spy 会看到大段 socket.recv 阻塞。
六、效果数据
用 wrk 压测，同样的机器和参数：
wrk -t8 -c200 -d60s --latency http://127.0.0.1:8000/api/v1/orders?user_id=10086
| 指标 | 优化前 | 优化后 | 提升 |
|---|---|---|---|
| P50 | 380ms | 62ms | 6.1x |
| P99 | 1210ms | 180ms | 6.7x |
| QPS | 320 | 1450 | 4.5x |
| 单请求 SQL 数 | 81 | 3 | 27x |
| MySQL CPU | 92% | 31% | — |
Redis 命中率稳定在 96.3%，L1 本地缓存命中率约 41%（热点用户集中）。
七、总结
这次调优给我的几点体会：
- 别猜，用数据说话。py-spy 10 分钟就能定位到 60% 的时间花在哪，比看代码猜半天强。
- N+1 是 API 性能的头号杀手。ORM 用起来爽，但一定要盯着 SQL 数量。
- 索引不是加得越多越好。组合索引要匹配
WHERE + ORDER BY，加错了反而拖慢写入。 - 缓存要设计好失效和穿透。不然压测时看着很美，上线就雪崩。
- 混合技术栈要隔离资源。FastAPI 和 Flask 共库这件事，本身就是个定时炸弹。
如果你的接口 P99 也超过 500ms，建议先跑一遍 py-spy，八成能发现问题。有问题欢迎评论区交流。
