一、问题背景
事情是这样的，我们有个订单查询接口 /api/v1/orders/summary，前端首页要调用它来展示用户的订单概览。逻辑不复杂：根据用户ID查出最近N条订单，再关联商品信息和物流状态返回。
上线初期用户量小，没人注意。直到运营搞了一次促销，用户量涨上来之后，监控开始报警：
- P99 延迟：2.3s
- QPS 到 120 左右就开始大量 504
- CPU 使用率：4 个 worker 全部打满
这个接口是首页首屏依赖的，挂了等于整个 App 首页白屏。必须马上处理。
二、环境与版本
先交代下环境，方便大家对号入座：
- Python 3.11.6
- FastAPI 0.109.2
- Uvicorn 0.27.1（4 workers，
--workers 4） - SQLAlchemy 2.0.25（ORM 模式）
- PostgreSQL 14.10
- Redis 7.2.4
- 部署：Docker + 4C8G 单机
压测工具用的 locust 2.20.0，脚本模拟用户携带 JWT 请求该接口。
三、方案设计：先定位，再动手
我的原则是：不要凭感觉优化。很多兄弟一上来就说"加缓存"，结果缓存了不该缓存的，反而引入一致性问题。
调优路线分三步：
- Profiling 定位：先用 cProfile 和 py-spy 找到时间花在哪
- 数据库层优化：解决 N+1 查询、加索引
- 缓存层优化：Redis 缓存热点数据
- 并发层调参：uvicorn worker 数量和 DB 连接池
四、核心实现
4.1 Profiling 定位瓶颈
先上 cProfile。写了个脚本直接调用接口的处理函数：
# profile_api.py
import cProfile
import pstats
import asyncio
from app.api.v1.orders import get_orders_summary
async def run():
# 模拟真实用户ID
await get_orders_summary(user_id=12345, limit=20)
if __name__ == "__main__":
profiler = cProfile.Profile()
profiler.enable()
asyncio.run(run())
profiler.disable()
stats = pstats.Stats(profiler).sort_stats("cumulative")
stats.print_stats(20)
输出关键片段：
ncalls tottime cumtime filename:lineno(function)
21 0.002 1.845 sqlalchemy/orm/loading.py:xxx(load)
420 0.031 1.712 psycopg2/extras.py:xxx(execute)
1 0.001 0.312 redis/client.py:xxx(get)
一眼看穿：420 次 SQL 执行，占了 1.7s。典型的 N+1 查询。
再用 py-spy 抓线上火焰图确认：
py-spy record -o profile.svg --pid --duration 30
火焰图里 psycopg2.execute 的栈又宽又高，实锤。
4.2 优化一：消灭 N+1 查询
原始代码长这样（简化版）：
# 优化前的错误写法
@router.get("/orders/summary")
async def get_orders_summary(user_id: int, limit: int = 20, db: AsyncSession = Depends(get_db)):
orders = await db.execute(
select(Order).where(Order.user_id == user_id).order_by(Order.created_at.desc()).limit(limit)
)
orders = orders.scalars().all()
result = []
for order in orders:
# 每次都单独查一次，20条订单 = 20次查询
items = await db.execute(select(OrderItem).where(OrderItem.order_id == order.id))
items = items.scalars().all()
logistics = await db.execute(select(Logistics).where(Logistics.order_id == order.id))
logistics = logistics.scalar_one_or_none()
result.append({
"order_id": order.id,
"items": [{"sku": i.sku, "qty": i.qty} for i in items],
"logistics_status": logistics.status if logistics else None,
})
return result
20 条订单 → 1 + 20 + 20 = 41 次查询。加上其他中间件查询，420 次也不奇怪。
改成 selectinload 预加载：
# 优化后
from sqlalchemy.orm import selectinload
@router.get("/orders/summary")
async def get_orders_summary(user_id: int, limit: int = 20, db: AsyncSession = Depends(get_db)):
stmt = (
select(Order)
.where(Order.user_id == user_id)
.order_by(Order.created_at.desc())
.limit(limit)
.options(
selectinload(Order.items),
selectinload(Order.logistics),
)
)
orders = (await db.execute(stmt)).scalars().all()
return [
{
"order_id": o.id,
"items": [{"sku": i.sku, "qty": i.qty} for i in o.items],
"logistics_status": o.logistics.status if o.logistics else None,
}
for o in orders
]
selectinload 会把关联查询合并成 WHERE order_id IN (...) 的一次查询，41 次 → 3 次。
4.3 优化二：加索引
查了下 orders 表的执行计划：
EXPLAIN ANALYZE SELECT * FROM orders WHERE user_id = 12345 ORDER BY created_at DESC LIMIT 20;
结果 Seq Scan，全表扫描 32 万行，耗时 180ms。
加复合索引：
CREATE INDEX CONCURRENTLY idx_orders_user_created
ON orders (user_id, created_at DESC);
CONCURRENTLY 是关键，避免锁表。加完后同样 SQL 走 Index Scan，耗时降到 1.2ms。
4.4 优化三：Redis 缓存
订单概览这种数据，用户短时间内不会变，缓存 30 秒完全够用。
import json
from redis.asyncio import Redis
redis_client = Redis.from_url("redis://localhost:6379/0", decode_responses=True)
CACHE_TTL = 30
@router.get("/orders/summary")
async def get_orders_summary(user_id: int, limit: int = 20, db: AsyncSession = Depends(get_db)):
cache_key = f"order:summary:{user_id}:{limit}"
cached = await redis_client.get(cache_key)
if cached:
return json.loads(cached)
# ... 上面的 DB 查询逻辑 ...
result = [...]
await redis_client.setex(cache_key, CACHE_TTL, json.dumps(result, default=str))
return result
注意几个坑：
json.dumps里的 datetime 要处理，default=str是简单方案- 缓存穿透：用户ID不存在时也缓存空列表，TTL 设短一点（5秒）
- 缓存雪崩：TTL 加随机抖动
30 + random.randint(0, 10)
4.5 优化四：Uvicorn 与连接池调参
原来的启动命令：
uvicorn app.main:app --workers 4 --host 0.0.0.0 --port 8000
问题在于这个接口是 IO 密集型（等 DB、等 Redis），4 个 worker 完全不够。改成 8 个 worker（CPU 核数 * 2）：
uvicorn app.main:app --workers 8 --host 0.0.0.0 --port 8000 --loop uvloop --http httptools
uvloop 和 httptools 是 uvicorn 的高性能实现，实测 QPS 能提升 15% 左右。
数据库连接池也要跟上：
engine = create_async_engine(
DATABASE_URL,
pool_size=20,
max_overflow=10,
pool_pre_ping=True,
pool_recycle=3600,
)
pool_size 要跟 worker 数量匹配。8 个 worker × 每个并发 3~4 个请求 = 大约需要 24~32 个连接，设置 pool_size=20, max_overflow=10 刚好。
五、踩坑与优化
踩了几个坑，记录一下：
坑 1：selectinload 顺序问题
一开始我写成了 .options(selectinload(Order.items)).limit(20)，结果 SQLAlchemy 警告 limit 在 join 场景下可能不准确。正确姿势是先 limit 再 options（链式调用顺序不影响语义，但要注意 joinedload 和 limit 同时用会出问题，selectinload 才是安全的）。
坑 2：Redis 连接没复用
一开始每次请求都 Redis.from_url()，连接开销比查询还大。改成模块级单例后，Redis 平均耗时从 8ms 降到 0.6ms。
坑 3：uvloop 与某些库不兼容
我们有个老的同步 SDK 用了 nest_asyncio，跟 uvloop 冲突，最后把那部分逻辑挪到单独线程池执行。
坑 4：缓存序列化开销
json.dumps 20 条订单大概 15KB，序列化耗时 2ms 左右。如果数据再大，建议用 orjson，实测快 3~5 倍：
import orjson
await redis_client.setex(cache_key, CACHE_TTL, orjson.dumps(result))
六、效果数据
用 locust 压测，200 并发用户，持续 5 分钟：
| 指标 | 优化前 | 优化后 | 提升 |
|---|---|---|---|
| P50 延迟 | 680ms | 32ms | 21x |
| P95 延迟 | 1.6s | 68ms | 23x |
| P99 延迟 | 2.3s | 89ms | 26x |
| QPS | 120 | 1420 | 11.8x |
| 错误率 | 8.3% | 0% | - |
| CPU 使用率 | 100% | 45% | - |
| 平均 SQL 次数/请求 | 420 | 3 | 140x |
线上灰度发布后观察一周，P99 稳定在 90~110ms，Redis 命中率 87%，DB 负载下降 70%。
七、总结
这次调优最大的心得：
- Profiling 是第一步，不是可选项。不看数据就动手，等于蒙眼开车。cProfile 看函数级、py-spy 看线上火焰图，两个配合用。
- N+1 是 ORM 的头号敌人。FastAPI + SQLAlchemy 的异步场景下尤其隐蔽，因为
await之间的查询看起来是"顺序执行"，实际是放大。 - 索引要按查询模式建。
(user_id, created_at DESC)这种复合索引比单列索引香太多，尤其配合ORDER BY ... LIMIT。 - 缓存不是万能的，但没缓存是万万不能的。热点数据 30 秒缓存 + 随机抖动，能扛住 90% 的流量。
- 参数调优是最后一步，但效果立竿见影。uvloop、httptools、连接池大小，这几个参数改完就见效。
如果你的接口也有类似问题，建议按这个顺序排查：Profiling → SQL 次数 → 索引 → 缓存 → 并发参数。基本能覆盖 90% 的性能问题。
有踩过类似坑的朋友欢迎评论区交流，特别是 SQLAlchemy 2.0 异步下的各种玄学问题，后面可以单独写一篇。
