一、问题背景
事情是这样的，我们有个内部数据服务平台，主要给运营后台提供订单维度的聚合查询。接口本身逻辑不复杂：根据用户ID和时间范围，返回订单列表 + 每单的商品明细 + 统计信息。
上线初期数据量小，没人关注性能。直到某天运营反馈"订单页要转好几秒"，我上去一看监控：这个接口P99已经飙到800ms，高峰期直接上1.5s，QPS超过50就开始请求堆积。
技术栈是FastAPI + SQLAlchemy + PostgreSQL + Redis，部署在4C8G的容器里，uvicorn单进程4 worker。说实话这套组合本身没问题，问题肯定出在代码上。
二、环境与版本
先把环境交代清楚，不同版本行为差异挺大的：
- Python 3.11.6
- FastAPI 0.109.0
- uvicorn 0.27.0（4 workers，
--loop uvloop） - SQLAlchemy 2.0.25（用的2.0风格ORM）
- asyncpg 0.29.0
- PostgreSQL 15.4
- Redis 7.2.3
- py-spy 0.3.14
- Locust 2.20.0
数据库表大致是：orders 表500万行，order_items 表2000万行，users 表20万行。
三、方案设计：先定位，再动手
我的原则是：没有profile数据的优化都是瞎猜。所以第一步不是改代码，而是搞清楚时间花在哪。
整体思路分三步：
- 定位：用py-spy采样 + cProfile函数级统计，找出热点函数
- 拆分：把接口耗时拆成DB时间、序列化时间、业务逻辑时间
- 优化：针对性地做查询优化、加索引、上缓存、调连接池
四、核心实现
4.1 用py-spy定位热点
生产环境不敢随便挂cProfile（开销大），先用py-spy采样看看：
# 找到uvicorn worker进程
ps aux | grep uvicorn
# 对PID 12345采样30秒，生成火焰图
py-spy record -o profile.svg --pid 12345 --duration 30 --rate 200
火焰图一出来就明显了：get_order_list 这个函数占了62% 的CPU时间，其中绝大部分又卡在 await session.execute() 上。
再用cProfile在本地复现，看函数级调用：
import cProfile
import pstats
async def benchmark():
# 模拟真实请求
for _ in range(100):
await get_order_list(user_id=12345, start="2024-01-01", end="2024-03-01")
cProfile.run("asyncio.run(benchmark())", "profile.out")
p = pstats.Stats("profile.out")
p.sort_stats("cumulative").print_stats(20)
结果很典型：
ncalls tottime cumtime function
100 0.012 78.4 get_order_list
5100 0.089 71.2 sqlalchemy.orm.loading
5000 0.045 68.9 asyncpg.protocol
100次请求花了78秒，平均780ms/次，其中SQLAlchemy的加载逻辑占了71秒。再看SQL日志，好家伙，每次请求发起了 51条SQL——典型的N+1问题。
4.2 罪魁祸首：N+1查询
原来的代码长这样（简化版）：
@app.get("/orders")
async def get_order_list(
user_id: int,
start: str,
end: str,
session: AsyncSession = Depends(get_session),
):
# 第1条：查订单
orders = await session.execute(
select(Order).where(
Order.user_id == user_id,
Order.created_at.between(start, end),
)
)
orders = orders.scalars().all()
result = []
for order in orders:
# 第N条：每单再查一次明细（致命）
items = await session.execute(
select(OrderItem).where(OrderItem.order_id == order.id)
)
result.append({
"id": order.id,
"amount": order.amount,
"items": items.scalars().all(),
})
return result
50个订单 → 1 + 50 = 51条SQL。每条SQL哪怕只有1ms，加上网络往返和ORM开销，轻松几百毫秒。
4.3 优化一：selectinload 消除N+1
SQLAlchemy 2.0的selectinload能把N+1变成2条SQL：
from sqlalchemy.orm import selectinload
@app.get("/orders")
async def get_order_list(
user_id: int,
start: str,
end: str,
session: AsyncSession = Depends(get_session),
):
stmt = (
select(Order)
.options(selectinload(Order.items)) # 关键：一次加载所有明细
.where(
Order.user_id == user_id,
Order.created_at.between(start, end),
)
.order_by(Order.created_at.desc())
.limit(100)
)
orders = (await session.execute(stmt)).scalars().all()
return [
{
"id": o.id,
"amount": o.amount,
"items": [{"sku": i.sku, "qty": i.qty} for i in o.items],
}
for o in orders
]
现在只有2条SQL。耗时从780ms降到290ms，但还不够。
4.4 优化二：加索引
看EXPLAIN ANALYZE发现orders表的查询走了全表扫描：
Seq Scan on orders (cost=0.00..145000.00 rows=50 width=...)
Filter: (user_id = 12345 AND created_at >= ... AND created_at <= ...)
Rows Removed by Filter: 4999950
扫了500万行才筛出50条，这谁顶得住。加复合索引：
CREATE INDEX CONCURRENTLY idx_orders_user_created
ON orders (user_id, created_at DESC);
CREATE INDEX CONCURRENTLY idx_order_items_order_id
ON order_items (order_id);
注意用CONCURRENTLY，生产环境加索引不锁表。加完后走Index Scan，查询从180ms降到8ms。
4.5 优化三：Redis缓存
查询接口的数据其实变化不频繁（订单完成后基本不动），很适合缓存。策略：
- Key：
orders:{user_id}:{start}:{end} - TTL：5分钟
- 序列化用orjson（比标准json快3倍）
import orjson
from redis.asyncio import Redis
redis = Redis.from_url("redis://localhost:6379/0", decode_responses=False)
async def get_order_list_cached(user_id, start, end, session):
cache_key = f"orders:{user_id}:{start}:{end}"
cached = await redis.get(cache_key)
if cached:
return orjson.loads(cached)
# ... 走DB查询逻辑 ...
result = [...]
await redis.setex(cache_key, 300, orjson.dumps(result))
return result
这里有个坑后面会说。缓存命中后P99直接降到45ms。
4.6 优化四：连接池调优
默认的连接池配置在高并发下会成瓶颈。调整：
engine = create_async_engine(
DATABASE_URL,
pool_size=20, # 默认5，调大
max_overflow=10, # 默认10
pool_pre_ping=True, # 防止拿到死连接
pool_recycle=1800, # 30分钟回收
echo=False,
)
pool_pre_ping会带来一点开销，但比"connection closed"报错强太多。
五、踩坑与优化
坑1：缓存击穿
上线第一天遇到缓存过期瞬间大量请求打到DB。加了个简单的互斥锁：
async def get_with_lock(key, fetch_func):
cached = await redis.get(key)
if cached:
return orjson.loads(cached)
lock_key = f"lock:{key}"
# SET NX EX 实现分布式锁
if await redis.set(lock_key, "1", nx=True, ex=10):
try:
data = await fetch_func()
await redis.setex(key, 300, orjson.dumps(data))
return data
finally:
await redis.delete(lock_key)
else:
# 没抢到锁，等一下重试
await asyncio.sleep(0.05)
return await get_with_lock(key, fetch_func)
坑2：uvicorn worker数不是越多越好
一开始我调到8个worker，结果QPS不升反降。原因是PG连接数被打满（8 workers × 30 pool_size = 240连接），PG默认max_connections才100。后来降到4 workers，pool_size降到20，反而更稳。
经验：worker数 × pool_size ≤ PG max_connections × 0.8。
坑3：selectinload 的 limit 陷阱
selectinload配合limit时要注意：SQLAlchemy会先对主查询limit，再按主查询的ID去IN查明细，这个行为是对的。但如果你在options里用了joinedload再配limit，就会因为JOIN导致行数膨胀，limit失效。所以一对多用selectinload，多对一用joinedload。
六、效果数据
用Locust压测，100并发持续3分钟：
| 指标 | 优化前 | 优化后 | 提升 |
|---|---|---|---|
| P50 | 420ms | 18ms | 23x |
| P95 | 780ms | 38ms | 20x |
| P99 | 820ms | 45ms | 18x |
| QPS | 52 | 620 | 12x |
| 错误率 | 3.2% | 0% | - |
| DB QPS | 2650 | 180 | 14x↓ |
DB QPS下降是因为缓存命中率到了87%。
CPU使用率从95%降到32%，内存稳定在1.2G。
单个请求的SQL数量：51 → 2（未命中缓存）/ 0（命中缓存）。
七、总结
这次优化让我再次确认几件事：
- 先profile再动手。我一开始以为瓶颈在序列化，差点去折腾orjson，结果火焰图直接打脸。
- N+1是API性能头号杀手。ORM用起来爽，但一定要用
selectinload/joinedload，别在循环里查DB。 - 索引不是加了就有用，要看
EXPLAIN确认走了正确的索引。复合索引的字段顺序很重要，等值条件在前，范围条件在后。 - 缓存是银弹也是炸弹。TTL、击穿、雪崩都要提前想好，别等出事再补。
- 连接池不是越大越好。超过DB承载能力反而拖慢整体。
最后，性能优化没有终点。这个接口从800ms到45ms，但业务量再涨10倍，可能又得考虑分库分表、读写分离了。先解决眼前的问题，别过度设计。
代码都脱敏过了，思路可以直接复用。有问题评论区聊。
参考：py-spy官方文档、SQLAlchemy 2.0 ORM加载策略、PostgreSQL索引使用指南
