一、问题背景
上个月接手了一个电商中台项目，技术栈是 FastAPI 0.110.0 + SQLAlchemy 2.0.29 + PostgreSQL 15.3 + Redis 7.2。上线后运营反馈后台"订单列表"页面加载特别慢，尤其是筛选"待发货"状态的时候，转圈能转到怀疑人生。
我先用 curl 简单测了一下：
curl -w "\n总耗时: %{time_total}s\n" -H "Authorization: Bearer xxx" \
"https://api.example.com/api/v1/orders?status=pending&page=1&size=20"
结果：
总耗时: 1.243s
这个接口逻辑很简单：查订单列表 + 关联查用户信息 + 关联查商品信息。按理说不该这么慢。更离谱的是，当我用 wrk 压测时（并发200，持续30秒），P99 直接飙到 1.8s，PostgreSQL 的 CPU 使用率冲到 95%，接口开始大量超时。
问题很明确：这个接口有性能瓶颈，而且瓶颈大概率在数据库层。但具体是慢查询、N+1、还是索引缺失？得用数据说话。
二、环境与版本
先说明一下环境，避免版本差异导致结论不可复现：
| 组件 | 版本 |
|---|---|
| Python | 3.11.8 |
| FastAPI | 0.110.0 |
| Uvicorn | 0.29.0 (workers=4) |
| SQLAlchemy | 2.0.29 |
| asyncpg | 0.29.0 |
| PostgreSQL | 15.3 |
| Redis | 7.2.4 |
| py-spy | 0.3.14 |
| wrk | 4.2.0 |
部署环境是 4C8G 的云服务器，数据库单独一台 8C16G。压测机跟服务同内网，避免网络抖动干扰。
三、定位瓶颈：py-spy + 慢查询日志
3.1 用 py-spy 抓火焰图
FastAPI 是异步框架，普通的 cProfile 对 async 代码支持不好，我直接用 py-spy 采样：
# 找到 uvicorn 进程
ps aux | grep uvicorn
# 采样 30 秒，生成火焰图
py-spy record -o profile.svg --pid 12345 --duration 30 --rate 100
同时用 wrk 打流量：
wrk -t4 -c200 -d30s --latency \
"http://127.0.0.1:8000/api/v1/orders?status=pending&page=1&size=20"
火焰图打开一看，asyncpg 的 _wait_for_query 占了 68% 的采样，也就是说大部分时间都在等数据库返回。这就排除了 Python 层的 CPU 密集问题，锁定在 SQL 层面。
3.2 打开 PostgreSQL 慢查询日志
在 postgresql.conf 里加上：
log_min_duration_statement = 200 # 超过200ms记录
log_line_prefix = '%t [%p] user=%u,db=%d '
重启后看日志，发现这个接口一次请求竟然发了 23 条 SQL！典型的 N+1 问题。其中一条查询：
SELECT * FROM order_items WHERE order_id = 10086;
单条只要 2ms，但一次请求要执行 20 次（每页20条订单），加上主查询和其他关联查询，累计就上去了。
3.3 用 EXPLAIN ANALYZE 看主查询
主查询是这样的：
SELECT * FROM orders
WHERE status = 'pending'
ORDER BY created_at DESC
LIMIT 20 OFFSET 0;
跑一下：
EXPLAIN (ANALYZE, BUFFERS) SELECT ...;
结果：
Seq Scan on orders (cost=0.00..45231.00 rows=198234 width=...)
(actual time=0.015..412.337 rows=20 loops=1)
Filter: (status = 'pending'::text)
Rows Removed by Filter: 198214
全表扫描，扫了 19 万行才筛出 20 条。orders 表有 20 万行数据，status 字段竟然没索引。这就是第二个瓶颈。
四、方案设计与核心实现
定位清楚了，优化分三步走：
- N+1 查询 → 用 SQLAlchemy 的
selectinload预加载 - 缺失索引 → 建复合索引
(status, created_at DESC) - 热点数据 → 加 Redis 缓存，TTL 30 秒
4.1 改造前的代码
# 改造前：N+1 重灾区
@app.get("/api/v1/orders")
async def list_orders(status: str, page: int = 1, size: int = 20, db: AsyncSession = Depends(get_db)):
offset = (page - 1) * size
result = await db.execute(
select(Order).where(Order.status == status)
.order_by(Order.created_at.desc())
.offset(offset).limit(size)
)
orders = result.scalars().all()
data = []
for o in orders:
# 循环里查两次，20条订单 = 40次查询
user = await db.get(User, o.user_id)
items = (await db.execute(
select(OrderItem).where(OrderItem.order_id == o.id)
)).scalars().all()
data.append({"order": o, "user": user, "items": items})
return data
4.2 优化后的代码
# 优化后：预加载 + 缓存 + 只查需要的字段
from sqlalchemy.orm import selectinload
from sqlalchemy import select
import json, hashlib
CACHE_TTL = 30
@app.get("/api/v1/orders")
async def list_orders(
status: str,
page: int = 1,
size: int = 20,
db: AsyncSession = Depends(get_db),
redis: Redis = Depends(get_redis),
):
# 1. 缓存 key（用参数拼 hash）
cache_key = "orders:" + hashlib.md5(
f"{status}:{page}:{size}".encode()
).hexdigest()
# 2. 先查缓存
cached = await redis.get(cache_key)
if cached:
return json.loads(cached)
# 3. 一次查询搞定关联（selectinload 会发 2 条 SQL 而非 40 条）
offset = (page - 1) * size
stmt = (
select(Order)
.options(
selectinload(Order.user),
selectinload(Order.items),
)
.where(Order.status == status)
.order_by(Order.created_at.desc())
.offset(offset)
.limit(size)
)
result = await db.execute(stmt)
orders = result.scalars().all()
data = [
{
"id": o.id,
"user": {"id": o.user.id, "name": o.user.name},
"items": [{"sku": i.sku, "qty": i.qty} for i in o.items],
"created_at": o.created_at.isoformat(),
}
for o in orders
]
# 4. 写缓存
await redis.setex(cache_key, CACHE_TTL, json.dumps(data))
return data
4.3 加索引
-- 复合索引，覆盖 WHERE + ORDER BY
CREATE INDEX CONCURRENTLY idx_orders_status_created
ON orders (status, created_at DESC);
-- 关联表的外键索引（之前也没建）
CREATE INDEX CONCURRENTLY idx_order_items_order_id
ON order_items (order_id);
CREATE INDEX CONCURRENTLY idx_orders_user_id
ON orders (user_id);
注意用 CONCURRENTLY，生产环境建索引不锁表。
五、踩坑与进一步优化
坑1：selectinload vs joinedload
一开始我用的是 joinedload，结果发现当订单有多个 items 时，主查询会产生笛卡尔积，返回行数爆涨，内存直接吃掉几百 MB。selectinload 是发第二条 WHERE order_id IN (...) 查询，更稳妥。记住：一对多用 selectinload，多对一用 joinedload。
坑2：缓存击穿
上线后发现热点 key 过期瞬间，几十个并发同时打到数据库。加了个简单的互斥锁：
lock_key = cache_key + ":lock"
if await redis.set(lock_key, "1", nx=True, ex=5):
try:
data = await query_db(...)
await redis.setex(cache_key, CACHE_TTL, json.dumps(data))
finally:
await redis.delete(lock_key)
else:
await asyncio.sleep(0.1)
cached = await redis.get(cache_key)
return json.loads(cached) if cached else []
坑3：连接池配置
默认 pool_size=5 在高并发下不够用，改成：
engine = create_async_engine(
DATABASE_URL,
pool_size=20,
max_overflow=10,
pool_pre_ping=True,
pool_recycle=3600,
)
坑4：COUNT 查询
分页前端要总数，SELECT COUNT(*) 在大表上很贵。我改成用 Redis 单独缓存总数，TTL 60 秒，或者干脆上 pg_class.reltuples 估算。
六、效果数据
优化前后用同样的 wrk 参数压测（4线程、200并发、30秒）：
| 指标 | 优化前 | 优化后 | 提升 |
|---|---|---|---|
| P50 | 680ms | 42ms | 16x |
| P99 | 1.82s | 80ms | 22x |
| QPS | 152 | 1108 | 7.3x |
| 单请求 SQL 数 | 23 | 2 | -91% |
| DB CPU | 95% | 22% | - |
| 错误率 | 3.7% | 0% | - |
单条接口的耗时从 1.243s 降到 78ms。火焰图上 _wait_for_query 的占比从 68% 掉到 15%。
七、总结
这次调优最深的感受是：不要凭感觉优化，一定要用数据定位。我一开始以为瓶颈在 FastAPI 的序列化，差点去换 orjson，结果 py-spy 一跑，全在等数据库。
三条经验：
- py-spy 是异步框架的 profiling 神器，比 cProfile 好用太多，生产环境也能直接跑。
- N+1 是 ORM 永恒的话题，SQLAlchemy 2.0 的
selectinload一定要用起来，别在循环里查数据库。 - 索引不是加了就行，要按
WHERE + ORDER BY的组合建复合索引，顺序错了照样全表扫。
最后提醒一句：缓存虽然香，但一致性要自己兜底。订单状态变更时记得主动删对应的缓存 key，否则用户会看到"幻觉订单"。这个坑我踩过，就不展开说了。
如果这篇文章帮你解决了一个慢接口，点个赞再走。有问题评论区聊。
