一、问题背景：接口慢到被运维盯上
上周四，运维同事直接在群里@我："/api/v1/orders/summary这个接口，生产环境P99耗时1.2秒，已经触发了告警阈值。"我打开Grafana一看，好家伙，这个接口每天被调用约80万次，P95延迟还在持续爬升。
这是个典型的报表聚合接口：根据用户ID查询其所有订单，按状态分组统计，还要附带每个订单关联的商品信息。代码是三个月前写的，当时数据量小，跑得飞快。现在订单表已经涨到500万行，问题就暴露了。
先说结论：经过两轮优化，P99从1.2s降到210ms，QPS从120提升到510，数据库CPU从85%降到30%。下面详细记录整个过程，包括我踩的坑。
二、环境与版本：先亮家底
本次调优涉及的软件版本如下，如果你环境类似，可以直接复现：
- Python 3.11.6
- FastAPI 0.104.1
- Uvicorn 0.24.0 (workers=4, loop="uvloop")
- SQLAlchemy 2.0.25 (async模式)
- PostgreSQL 15.3（连接池：asyncpg 0.29.0，池大小20）
- Redis 7.2（用于缓存，读取为主）
- 压测工具：wrk 4.2.0
服务器是4核8G的容器，宿主机是物理机（CPU型号不透露了，反正不是最差的那种）。压测命令统一用：
wrk -t4 -c100 -d30s --latency http://localhost:8000/api/v1/orders/summary?user_id=12345
这里-t4表示4个线程，-c100保持100个并发连接，-d30s持续30秒。注意我用的是固定user_id，避免缓存穿透带来的干扰。
三、第一步：Profiling定位瓶颈，别瞎猜
我见过太多人一上来就加缓存、加索引，结果瓶颈根本不在这。我的习惯是：先量化，再动手。
先看接口代码原貌（简化版）：
# app/routers/orders.py
from fastapi import APIRouter, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, func
from app.db import get_db
from app.models import Order, OrderItem, Product
router = APIRouter()
@router.get("/orders/summary")
async def get_order_summary(user_id: int, db: AsyncSession = Depends(get_db)):
# 查询该用户的所有订单
orders = await db.execute(
select(Order).where(Order.user_id == user_id)
)
orders = orders.scalars().all()
# 对每个订单，查询其商品明细
result = {"total": 0, "by_status": {}, "items": []}
for order in orders:
result["total"] += order.amount
status = order.status
result["by_status"][status] = result["by_status"].get(status, 0) + 1
# N+1查询：每个订单查一次商品表
items = await db.execute(
select(OrderItem).where(OrderItem.order_id == order.id)
)
for item in items.scalars().all():
product = await db.get(Product, item.product_id) # 又一个N+1
result["items"].append({
"order_id": order.id,
"product_name": product.name,
"quantity": item.quantity
})
return result
这段代码有两个明显的N+1问题：订单查完再查OrderItem，OrderItem查完再查Product。如果用户有100个订单，每个订单平均5个商品，那就是100次OrderItem查询 + 500次Product主键查询。异步接口虽然不阻塞，但数据库往返次数太多，延迟全花在等待上了。
用cProfile跑一次（注意cProfile对async支持不太好，我改用py-spy）：
py-spy record -o profile.svg --pid $(pgrep -f uvicorn) --duration 30
火焰图导出来后，一眼就看出问题：数据库查询耗时占了总耗时的82%，其中OrderItem和Product的查询占了绝大部分。另外发现一个隐藏问题：result["by_status"]这个字典每次请求都重新计算，如果用户历史订单有1000条，这个循环也要跑1000次，虽然不慢，但没必要。
四、数据库查询优化：合并N+1
定位到问题后，第一刀砍向N+1。用SQLAlchemy 2.0的selectinload一次性加载关联对象，避免逐条查询。
优化后的代码：
# app/routers/orders.py (优化版)
from sqlalchemy.orm import selectinload
@router.get("/orders/summary")
async def get_order_summary(user_id: int, db: AsyncSession = Depends(get_db)):
# 一次性查询订单+关联的OrderItem+Product，消除N+1
result = await db.execute(
select(Order)
.options(
selectinload(Order.items).selectinload(OrderItem.product)
)
.where(Order.user_id == user_id)
)
orders = result.scalars().all()
# 内存中聚合
result = {"total": 0, "by_status": {}, "items": []}
for order in orders:
result["total"] += order.amount
result["by_status"][order.status] = result["by_status"].get(order.status, 0) + 1
for item in order.items:
result["items"].append({
"order_id": order.id,
"product_name": item.product.name,
"quantity": item.quantity
})
return result
selectinload会生成一条额外的WHERE order_id IN (...)的批量查询，替代原来的逐条查询。这里有个坑：selectinload只对sessions有效，如果你用了lazy="raise"的模型配置，必须显式指定。另外，如果你返回的是ORM对象并且序列化成JSON，记得用exclude排除不需要的字段，否则会把整个关联对象都序列化出去。
压测对比（wrk结果）：
| 指标 | 优化前 | 优化后 | 变化 |
|---|---|---|---|
| P50 | 620ms | 180ms | -71% |
| P99 | 1.2s | 340ms | -72% |
| QPS | 120 | 320 | +167% |
| 数据库CPU | 85% | 55% | -35% |
这一刀下去，效果已经很明显了。但P99还在340ms，不够好。我继续看火焰图，发现数据库查询占比降到了45%，但序列化和JSON构建占比升到了30%（因为返回的数据量变大了）。而且如果用户订单量特别大（比如500个订单，50个商品），单次请求的数据量还是很大。
五、缓存策略：给热点数据加个罩子
考虑到这个接口是按用户维度聚合的，而且用户的订单数据不会频繁变化（一天最多变几次），非常适合引入Redis缓存。但缓存有几个坑要注意：
- 缓存穿透：如果user_id不存在，每次都会打到数据库。解决：缓存空结果，TTL设短一点（比如60秒）。
- 缓存雪崩：如果所有key同时过期，瞬间请求全部打到数据库。解决：TTL加随机抖动。
- 数据一致性：订单状态变更时，必须主动删除缓存。
我的设计如下（简化版）：
# app/services/cache.py
import json
import random
import redis.asyncio as aioredis
redis_client = aioredis.from_url("redis://localhost:6379/0", decode_responses=True)
CACHE_TTL_BASE = 300 # 5分钟基础TTL
async def get_order_summary_cached(user_id: int, db: AsyncSession):
cache_key = f"order_summary:v1:{user_id}"
# 尝试从缓存读取
cached = await redis_client.get(cache_key)
if cached:
return json.loads(cached)
# 缓存未命中，查数据库
result = await db.execute(
select(Order)
.options(selectinload(Order.items).selectinload(OrderItem.product))
.where(Order.user_id == user_id)
)
orders = result.scalars().all()
# 构建聚合结果
summary = {"total": 0, "by_status": {}, "items": []}
for order in orders:
summary["total"] += order.amount
summary["by_status"][order.status] = summary["by_status"].get(order.status, 0) + 1
for item in order.items:
summary["items"].append({
"order_id": order.id,
"product_name": item.product.name,
"quantity": item.quantity
})
# 写入缓存，TTL加随机抖动避免雪崩
ttl = CACHE_TTL_BASE + random.randint(0, 120)
await redis_client.set(cache_key, json.dumps(summary), ex=ttl)
return summary
踩坑记录：
- 坑1：一开始我直接把ORM对象
json.dumps，结果报TypeError: Object of type Order is not JSON serializable。必须手动构建dict或使用Pydantic的model_validate。 - 坑2：Redis连接池默认是100个连接，但我们压测100并发时，每个请求都抢连接，导致部分请求等待连接超时。调大
max_connections=200解决。 - 坑3：订单状态更新时，我忘了删缓存，导致数据显示不一致。后来在写操作的地方加了
await redis_client.delete(cache_key)。如果你有多个写入口，建议用pub/sub广播删除事件。
六、最终压测数据与总结
加上缓存后，再次压测（注意：第一次压测会有缓存穿透，第二次开始才是真实数据）：
| 指标 | 原始版本 | 优化查询 | +Redis缓存 |
|---|---|---|---|
| P50 | 620ms | 180ms | 45ms |
| P99 | 1.2s | 340ms | 210ms |
| QPS | 120 | 320 | 510 |
| 数据库CPU | 85% | 55% | 30% |
| Redis内存 | - | - | 约120MB |
最后补充两点：
- 压测时一定要看冷热数据，我上面记录的是热数据（缓存命中）的结果。如果缓存全部过期（比如重启Redis后），性能会退化到第二步的水平。所以缓存命中率这个指标要监控起来。
- 异步接口的瓶颈经常不在代码逻辑，而在数据库往返次数。如果你看到火焰图里
await时间占比极高，大概率是IO问题，别急着加并发。
这次调优最大的收获是：先花30分钟做Profiling，比花3小时猜瓶颈强得多。工具层面，py-spy对异步代码的支持比cProfile好太多，强烈推荐。另外，SQLAlchemy 2.0的selectinload用起来比老版本的joinedload顺手，生成SQL更可控。
如果你也在优化FastAPI接口，建议按这个顺序排查：数据库查询次数（N+1）→ 索引是否命中 → 数据序列化开销 → 缓存策略。希望这篇记录对你有用。