一、问题背景
先说结论：一个用 FastAPI 写的订单查询接口 /api/v1/orders/{user_id}，在线上环境 P95 响应时间 1200ms，QPS 只有 45 左右，一到晚高峰就开始大量超时告警。
这个接口逻辑本身不复杂：根据 user_id 查最近 50 条订单，每条订单再带上商品信息、物流状态。之前一直是能跑的，但随着单用户订单量增长（有些老用户订单上千条），接口越来越慢。
我接手的时候，运维给的监控数据是：
- 平均响应时间：680ms
- P95：1200ms
- P99：2400ms
- 单实例 QPS：45
- CPU 使用率：35%（不高，说明在等 IO）
- 数据库连接数：经常打满到 100
CPU 不高但响应慢，典型的 IO 等待型瓶颈。下面记录完整的排查和优化过程。
二、环境与版本
先把环境列清楚，避免版本差异导致的结论不一致：
- Python 3.11.6
- FastAPI 0.109.2
- Uvicorn 0.27.0（workers=4，单机 4 核 8G）
- SQLAlchemy 2.0.25（async 模式，asyncpg 0.29.0）
- PostgreSQL 14.10
- Redis 7.2.3
- 压测工具：wrk 4.2.0 + locust 2.20.0
顺便说一句，项目里还有几个老的 Flask 接口（Flask 2.3.3 + Gunicorn），这次也一并做了对比优化，后面会提到。
三、方案设计：先定位，再动手
我的原则是：没有 profiling 数据的优化都是瞎猜。所以第一步不是改代码，而是搞清楚时间花在哪。
排查分三步：
- 用 py-spy 抓火焰图，看 Python 层面哪里卡住
- 用 cProfile 做单请求级别的函数耗时分析，定位到具体函数
- 打开 SQLAlchemy 的 echo，看实际执行了多少条 SQL
定位手段确定后，优化方向大致是三个：
- 数据库查询：N+1 问题、缺索引、一次性拉取过多字段
- 缓存：热点数据用 Redis 缓存，减少重复查询
- 并发模型：把能并行的 IO 并行化，调整连接池
四、核心实现
4.1 用 py-spy 抓火焰图
线上环境不方便改代码，py-spy 可以直接 attach 到进程，零侵入：
# 安装
pip install py-spy==0.3.14
# 抓 30 秒火焰图
py-spy record -o profile.svg --pid --duration 30 --rate 100
火焰图一出来就发现问题了：大量时间花在 asyncpg 的 fetch 上，而且调用栈里同一个 get_product_by_id 函数被反复调用了几十次。这就是典型的 N+1。
4.2 cProfile 单请求分析
为了拿到精确数字，我写了个脚本单独跑一次请求：
import cProfile
import pstats
import asyncio
from app.api.orders import get_user_orders
async def main():
profiler = cProfile.Profile()
profiler.enable()
# 模拟请求，user_id=12345 有 800 条订单
await get_user_orders(user_id=12345, limit=50)
profiler.disable()
stats = pstats.Stats(profiler).sort_stats("cumulative")
stats.print_stats(20)
asyncio.run(main())
输出里最扎眼的一条：
ncalls tottime cumtime function
51 0.002 1.180 get_product_by_id
50 条订单触发了 51 次商品查询，每次约 23ms，光这一项就 1.18 秒。问题确认。
4.3 优化一：消除 N+1
原始代码长这样（简化版）：
# 优化前：N+1
@router.get("/orders/{user_id}")
async def get_user_orders(user_id: int, limit: int = 50, db: AsyncSession = Depends(get_db)):
result = await db.execute(
select(Order).where(Order.user_id == user_id).order_by(Order.created_at.desc()).limit(limit)
)
orders = result.scalars().all()
items = []
for order in orders:
# 每条订单单独查商品，N+1 元凶
product = await db.execute(select(Product).where(Product.id == order.product_id))
product = product.scalar_one()
items.append({"order_id": order.id, "product": product.name, "amount": order.amount})
return {"items": items}
改成用 selectinload 一次性预加载关联对象：
# 优化后：预加载 + 只取需要的字段
from sqlalchemy.orm import selectinload
@router.get("/orders/{user_id}")
async def get_user_orders(user_id: int, limit: int = 50, db: AsyncSession = Depends(get_db)):
stmt = (
select(Order)
.options(selectinload(Order.product))
.where(Order.user_id == user_id)
.order_by(Order.created_at.desc())
.limit(limit)
)
result = await db.execute(stmt)
orders = result.scalars().all()
return {
"items": [
{"order_id": o.id, "product": o.product.name, "amount": o.amount}
for o in orders
]
}
这一改，SQL 从 51 条降到 2 条，接口耗时直接从 1180ms 掉到 210ms。
4.4 优化二：加索引
Order.user_id 上原本没有索引，800 条订单的全表扫描也要几十毫秒。加上复合索引：
CREATE INDEX CONCURRENTLY idx_orders_user_created
ON orders (user_id, created_at DESC);
注意用 CONCURRENTLY，线上大表加索引不锁表。
4.5 优化三：Redis 缓存
这个接口有个特点：订单列表在短时间内变化不大，但被高频访问（用户刷新页面、App 轮询）。所以对第一页做缓存很划算：
import json
import redis.asyncio as redis
redis_client = redis.Redis(host="127.0.0.1", port=6379, db=0, decode_responses=True)
CACHE_TTL = 60 # 秒
@router.get("/orders/{user_id}")
async def get_user_orders(user_id: int, limit: int = 50, page: int = 1, db: AsyncSession = Depends(get_db)):
cache_key = f"orders:{user_id}:{limit}:{page}"
# 只缓存第一页，命中率高
if page == 1:
cached = await redis_client.get(cache_key)
if cached:
return json.loads(cached)
stmt = (
select(Order)
.options(selectinload(Order.product))
.where(Order.user_id == user_id)
.order_by(Order.created_at.desc())
.limit(limit)
.offset((page - 1) * limit)
)
result = await db.execute(stmt)
orders = result.scalars().all()
data = {
"items": [
{"order_id": o.id, "product": o.product.name, "amount": o.amount}
for o in orders
]
}
if page == 1:
await redis_client.setex(cache_key, CACHE_TTL, json.dumps(data))
return data
这里有个坑：写订单时要主动删缓存，否则用户下单后刷不出来。用 redis_client.delete(f"orders:{user_id}:*") 配合 SCAN 删（KEYS 命令线上禁用）。
4.6 优化四：连接池调优
Uvicorn 4 workers，每个 worker 默认连接池 pool_size=5，总共 20 个连接，根本不够。调整 SQLAlchemy 的 pool：
from sqlalchemy.ext.asyncio import create_async_engine
engine = create_async_engine(
"postgresql+asyncpg://user:pass@localhost:5432/orders",
pool_size=20,
max_overflow=10,
pool_pre_ping=True,
pool_recycle=1800,
echo=False,
)
20 * 4 = 80 个连接，配合 PostgreSQL 的 max_connections=200，充裕。
4.7 Flask 老接口的对比优化
项目里还有个 Flask 的 /api/v1/products/{id} 接口，用的是同步 SQLAlchemy + Gunicorn（4 workers，sync worker）。这个接口 QPS 只有 30，优化思路类似：
- 换成
geventworker：gunicorn -k gevent -w 4 --worker-connections 1000 - 加 Flask-Caching + Redis：
CACHE_TYPE=redis，CACHE_DEFAULT_TIMEOUT=120 - 商品查询用
joinedload替代懒加载
改完后 QPS 从 30 提升到 180，提升幅度不如 FastAPI 那边，主要是 sync worker 的并发模型限制。
五、踩坑与优化
几个踩过的坑，记录一下：
坑 1：selectinload 不适用于多对一。 一开始对 Order.product 用了 selectinload，虽然能work，但实际上是发了两条 SQL。多对一关系用 joinedload 更合适，能合并成一条 JOIN。后来改成了 joinedload，又省了一次往返。
坑 2：JSON 序列化成了新瓶颈。 缓存里存的是 JSON 字符串，每次 json.loads 反序列化 50 条记录要 8ms 左右。数据量大时可以考虑用 orjson 替代标准库。
坑 3：缓存击穿。 热门用户缓存过期瞬间，大量请求同时打到数据库。加了简单的互斥锁（redis.setnx）做单飞，或者对热点 key 用随机 TTL 打散。
坑 4：连接池参数不是越大越好。 一开始 pool_size 调到 50，结果 PostgreSQL 连接开销反而拖慢了。最终 20 是压测出来的甜点值。
六、效果数据
用 wrk 在同样硬件上压测，对比优化前后（每个配置跑 3 次取中位数）：
| 指标 | 优化前 | 优化后 | 提升 |
|---|---|---|---|
| 平均响应时间 | 680ms | 95ms | 7.2x |
| P95 | 1200ms | 180ms | 6.7x |
| P99 | 2400ms | 320ms | 7.5x |
| QPS（单实例） | 45 | 320 | 7.1x |
| 数据库 QPS | 2300 | 180 | 下降 92% |
| 缓存命中率 | - | 78% | - |
压测命令：
wrk -t8 -c100 -d30s --latency \
"http://localhost:8000/api/v1/orders/12345?limit=50&page=1"
线上灰度一周后，接口超时告警从每天 200+ 降到 0，数据库连接数峰值从 100 降到 35。CPU 使用率反而升到 55%，因为处理能力上去了。
七、总结
这次优化下来，几点体会：
- profiling 优先于优化。 py-spy + cProfile 组合，10 分钟就能定位到瓶颈，比拍脑袋强一百倍。
- N+1 是老生常谈，但真的常见。 只要用了 ORM，就要警惕这个，
selectinload/joinedload是标配。 - 缓存不是银弹，要注意失效和击穿。 缓存 key 设计、TTL、主动失效三件事想清楚再上。
- 连接池参数要压测。 不同硬件、不同数据库配置下的最优值不一样，拍脑袋设定迟早出事。
- FastAPI 的异步优势要用起来。 Flask 那边即使换了 gevent，并发能力还是差一截，新项目建议直接上 FastAPI。
代码里的完整示例可以直接跑，但记得把数据库连接、Redis 地址换成你自己的。有问题的欢迎评论区讨论。
