一、问题背景：一个"看起来很简单"的接口
事情是这样的，我们有个订单列表接口 GET /api/v1/orders，功能非常朴素：按用户 ID 分页查询订单，每条订单带上商品信息和用户昵称。逻辑上就是几个查询拼一拼，代码不到 40 行。
上线初期没什么量，谁也没在意。直到运营做了一次促销活动，流量涨了大概 5 倍，监控开始报警：
- P95 延迟：820ms
- P99 延迟：1.6s
- 错误率：4.7%（主要是客户端超时重试）
- 单实例 QPS 到 210 左右就打满
用户投诉"订单页转圈圈"，老板在群里 @ 了三次。行吧，开始干活。
二、环境与版本
先把环境交代清楚，不同版本的坑差别很大：
- Python 3.11.6
- FastAPI 0.110.0
- Uvicorn 0.27.1（
--workers 4） - SQLAlchemy 2.0.28
- asyncpg 0.29.0
- PostgreSQL 14.10
- Redis 7.2
- 压测工具：wrk 4.2.0 + locust 2.20
- 部署：4C8G 容器，单实例
三、方案设计：先测，别猜
我见过太多人一上来就说"加缓存吧"，然后缓存加了一堆，慢的地方一点没变。性能优化的第一步永远是定位，不是动手。
我的排查路线：
- 压测复现：用 wrk 在本地稳定复现 800ms 的 P95
- 应用层 profiling：cProfile 看函数耗时分布，py-spy 抓火焰图
- SQL 层 profiling：打开 SQLAlchemy 慢查询日志 + PostgreSQL
pg_stat_statements - 确定瓶颈后针对性优化：ORM 加载策略 → 索引 → 缓存
- 回归压测：对比数据
四、核心实现：从定位到优化
4.1 压测复现
先用 wrk 稳定复现问题：
wrk -t8 -c100 -d60s --latency \
"http://127.0.0.1:8000/api/v1/orders?user_id=12345&page=1&size=20"
初始结果：
Latency Distribution
50% 312.45ms
75% 589.12ms
90% 743.88ms
99% 1.62s
Requests/sec: 208.34
4.2 py-spy 抓火焰图
这是我认为最省事的 profiling 方式，不需要改代码、不需要重启：
py-spy record -o profile.svg --pid $(pgrep -f uvicorn | head -1) --duration 30
火焰图上一眼就能看出来，sqlalchemy/orm/loading.py 的时间占比接近 70%，而且是很多次小查询堆叠，典型的 N+1。
同时打开 SQLAlchemy 的 echo，日志刷屏：
engine = create_async_engine(DATABASE_URL, echo=True)
打印出来是这样的：
SELECT * FROM orders WHERE user_id=$1 LIMIT 20; -- 1 次
SELECT * FROM products WHERE id=$1; -- 20 次
SELECT * FROM users WHERE id=$1; -- 20 次
一个请求 41 次 SQL，每次 15-20ms 网络往返，加起来 700ms+，完全对上了。
4.3 优化一：干掉 N+1
原来的代码（简化版）：
@router.get("/orders")
async def list_orders(user_id: int, page: int = 1, size: int = 20):
async with AsyncSession(engine) as session:
result = await session.execute(
select(Order)
.where(Order.user_id == user_id)
.offset((page - 1) * size)
.limit(size)
)
orders = result.scalars().all()
items = []
for order in orders: # 每次循环都查一次数据库
product = await session.get(Product, order.product_id)
user = await session.get(User, order.user_id)
items.append({
"id": order.id,
"amount": order.amount,
"product_name": product.name,
"user_name": user.nickname,
})
return items
改成 selectinload 一次性把关联数据拉进来：
from sqlalchemy.orm import selectinload
@router.get("/orders")
async def list_orders(user_id: int, page: int = 1, size: int = 20):
async with AsyncSession(engine) as session:
stmt = (
select(Order)
.options(
selectinload(Order.product),
selectinload(Order.user),
)
.where(Order.user_id == user_id)
.order_by(Order.created_at.desc())
.offset((page - 1) * size)
.limit(size)
)
result = await session.execute(stmt)
orders = result.scalars().all()
return [
{
"id": o.id,
"amount": o.amount,
"product_name": o.product.name,
"user_name": o.user.nickname,
}
for o in orders
]
SQL 从 41 条降到 3 条：
SELECT * FROM orders WHERE user_id=$1 ORDER BY created_at DESC LIMIT 20;
SELECT * FROM products WHERE id IN ($1, $2, ...);
SELECT * FROM users WHERE id IN ($1, $2, ...);
压测结果：P95 从 820ms → 340ms。有进步，但还不够。
4.4 优化二：加索引
打开 pg_stat_statements 看慢查询：
SELECT query, calls, mean_exec_time, rows
FROM pg_stat_statements
ORDER BY mean_exec_time DESC
LIMIT 10;
发现 orders 表上 user_id + created_at 的复合索引没建，全表扫描 12 万行。补上：
CREATE INDEX CONCURRENTLY idx_orders_user_created
ON orders (user_id, created_at DESC);
顺带把 products.id、users.id 主键索引确认了一遍（这俩本来就有）。
这一刀下去，P95 从 340ms → 180ms。
4.5 优化三：加缓存
180ms 其实已经能接受了，但这个接口的读取特征非常明显：
- 数据几乎不变（订单创建后很少更新）
- 同一个 user_id 反复请求（用户刷新、翻页）
- 允许秒级延迟
典型缓存友好场景。上 Redis，缓存粒度选"用户 + 页码"：
import json
import redis.asyncio as redis
redis_client = redis.from_url("redis://localhost:6379/0", decode_responses=True)
CACHE_TTL = 60 # 秒
@router.get("/orders")
async def list_orders(user_id: int, page: int = 1, size: int = 20):
cache_key = f"orders:{user_id}:{page}:{size}"
cached = await redis_client.get(cache_key)
if cached:
return json.loads(cached)
async with AsyncSession(engine) as session:
stmt = (
select(Order)
.options(selectinload(Order.product), selectinload(Order.user))
.where(Order.user_id == user_id)
.order_by(Order.created_at.desc())
.offset((page - 1) * size)
.limit(size)
)
result = await session.execute(stmt)
orders = result.scalars().all()
data = [
{
"id": o.id,
"amount": o.amount,
"product_name": o.product.name,
"user_name": o.user.nickname,
}
for o in orders
]
await redis_client.setex(cache_key, CACHE_TTL, json.dumps(data))
return data
写入侧记得失效缓存：
async def create_order(...):
# ... 创建订单逻辑
# 用 SCAN 清理该用户的所有分页缓存
async for key in redis_client.scan_iter(match=f"orders:{user_id}:*"):
await redis_client.delete(key)
压测结果：命中缓存的请求 P95 45ms，未命中的 180ms 左右。整体 P95 落在 60ms 以内。
五、踩坑与优化
坑 1：selectinload vs joinedload 选错
一开始我用了 joinedload，结果分页 LIMIT 20 因为 JOIN 产生笛卡尔积，实际捞回来的行数爆炸。selectinload 会先查主表再 IN 查关联表，分页语义才正确。记住：分页场景用 selectinload，一对一且明确唯一时才能用 joinedload。
坑 2：缓存雪崩
TTL 都是 60s，同一时刻大量 key 一起过期，瞬间把数据库打回原形。解决：TTL 加随机抖动：
import random
ttl = CACHE_TTL + random.randint(-10, 10)
await redis_client.setex(cache_key, ttl, json.dumps(data))
坑 3：selectinload 的 IN 列表太大
如果一页 20 条还没事，但如果有接口一页拉 1000 条，IN 里 1000 个 ID，asyncpg 的参数绑定会很难看。可以用 selectinload(...).selectin_polymorphic 或干脆限制单页 size ≤ 100。
坑 4：py-spy 采样看不到 await 挂起
py-spy 对纯异步代码的等待时间展示不直观，如果哪一步卡在 IO，火焰图上不一定看得出来。配合 loguru 打点，或者用 aiomonitor 看协程状态更靠谱。
坑 5：uvicorn worker 数
4C8G 的机器我一开始上 8 个 worker，反而更慢。原因是数据库连接池每个 worker 一份，8 worker × pool_size 20 = 160 连接，PostgreSQL 的 max_connections 默认 100，直接打满连接。后来改成 4 worker × pool_size 10 = 40 连接，稳定很多。
六、效果数据
同一台机器，同一份压测脚本，wrk -t8 -c100 -d60s：
| 阶段 | P50 | P95 | P99 | QPS | 错误率 |
|---|---|---|---|---|---|
| 优化前 | 312ms | 820ms | 1.62s | 208 | 4.7% |
| 干掉 N+1 | 128ms | 340ms | 620ms | 512 | 0.8% |
| + 索引 | 62ms | 180ms | 320ms | 890 | 0.1% |
| + Redis 缓存 | 18ms | 45ms | 92ms | 1420 | 0% |
缓存命中率稳定在 87% 左右（60s TTL，用户行为集中在短时间内）。
顺便说下，这个优化过程大概花了两个下午，其中一半时间在压测和验证，真正改代码的时间不到 2 小时。性能优化就是这样，80% 的时间花在定位，20% 的时间花在动手。
七、总结
回头看，这个案例里没有什么"黑科技"，就是三个经典手段按顺序执行：
- profiling 定位：py-spy + SQLAlchemy echo + pg_stat_statements，三个工具组合拳，5 分钟就能把瓶颈锁定
- ORM 查询优化：N+1 是 ORM 性能杀手，
selectinload是解药，但要分清使用场景 - 数据库索引：80% 的慢查询因为没有对的复合索引，
EXPLAIN ANALYZE是最好的朋友 - 缓存：不是万能药，但在读多写少 + 容忍秒级延迟的场景下，收益巨大
最后一句忠告：上线前压一遍，比上线后救火便宜一百倍。如果你手上也有慢接口，不妨从 py-spy 开始，先看真问题在哪。
