一、问题背景：监控告警先于用户投诉
周一早上10:23，Grafana告警弹出：orders_api 的P95延迟超过2000ms。我打开Kibana日志，看到大量超时记录——客户端设置的3秒超时，已经有0.3%的请求被中断。更糟糕的是，RDS的CPU使用率持续在95%以上，慢查询日志每分钟刷出几十条。
这个API是订单列表接口，前端分页展示用户的历史订单，每次请求返回20条订单摘要。逻辑本身不复杂：查订单表 → 关联订单项 → 关联商品快照 → 关联店铺信息。但线上就是慢，而且越到下午越明显。
二、环境与版本：先亮出底牌
Python: 3.11.4
FastAPI: 0.104.1 (基于Starlette 0.27.0)
Uvicorn: 0.23.2 (workers=4, loop=asyncio)
SQLAlchemy: 2.0.21 (asyncpg驱动)
PostgreSQL: 15.3 (RDS, db.r6g.xlarge, 4vCPU/32GB)
Redis: 7.0.12 (elasticache, cache.r6g.large)
压测工具: wrk 4.2.0
压测命令统一为：wrk -t8 -c200 -d60s --latency http://localhost:8000/api/v1/orders?user_id=12345&page=1&size=20
三、方案设计：先用Profile说话，不靠猜
我见过太多人一上来就加缓存，结果缓存了错误的数据，或者缓存了本可以不查的数据。我的原则是：先用Profiler定位，再谈优化。
3.1 第一步：cProfile快速定位
FastAPI的异步特性让cProfile有些水土不服，但我们可以把核心逻辑提出来同步执行。我写了一个profile脚本，直接复现API的调用链：
import cProfile
import pstats
import asyncio
from app.services.order_service import get_orders
async def profile_api():
# 模拟真实请求参数
user_id = 12345
page, size = 1, 20
# 直接调用Service层，跳过HTTP层
result = await get_orders(user_id, page, size)
return result
if __name__ == "__main__":
profiler = cProfile.Profile()
profiler.enable()
asyncio.run(profile_api())
profiler.disable()
stats = pstats.Stats(profiler)
stats.sort_stats('cumulative')
stats.print_stats(30) # 打印前30行耗时
输出结果（节选关键部分）：
ncalls tottime percall cumtime percall filename:lineno(function)
1 0.002 0.002 2.180 2.180 order_service.py:56(get_orders)
20 0.001 0.000 1.542 0.077 order_service.py:89(_get_order_items)
20 0.001 0.000 0.986 0.049 order_service.py:112(_get_product_snapshots)
20 0.001 0.000 0.421 0.021 order_service.py:135(_get_shop_info)
60 0.864 0.014 0.864 0.014 asyncpg/protocol/protocol.pyx:... (execute)
看到了吗？_get_order_items 被调用了20次——每个订单一次。这就是典型的N+1查询。总耗时2.18秒，其中1.54秒花在循环查询订单项上。
3.2 第二步：py-spy生成火焰图（生产环境实测）
cProfile只能在测试环境跑，生产环境用py-spy：
# 在服务器上执行，不重启进程
py-spy record --pid 12345 -o flamegraph.svg --duration 30
火焰图清晰显示：_get_order_items 和 _get_product_snapshots 占用了超过70%的CPU时间，而且大部分时间花在数据库往返上（asyncpg的socket读取）。
四、核心实现：数据库查询优化
4.1 用selectinload干掉N+1
原代码用的是lazy="select"，SQLAlchemy 2.0虽然默认改成了lazy="raise"，但我们的代码里显式写了lazy="select"。改成selectinload：
# 改造前（简化版）
async def get_orders(user_id: int, page: int, size: int):
stmt = select(Order).where(Order.user_id == user_id)\
.options(lazyload(Order.items)) # 这是坑
result = await session.execute(stmt)
orders = result.scalars().all()
# N+1查询：每个订单单独查items
for order in orders:
items = await session.execute(
select(OrderItem).where(OrderItem.order_id == order.id)
)
order._items = items.scalars().all()
return orders
# 改造后：一条SQL搞定
async def get_orders(user_id: int, page: int, size: int):
stmt = select(Order)\
.where(Order.user_id == user_id)\
.options(
selectinload(Order.items).selectinload(OrderItem.product_snapshot),
selectinload(Order.shop)
)\
.limit(size).offset((page-1)*size)
result = await session.execute(stmt)
return result.scalars().all()
selectinload会生成两条SQL：第一条查订单，第二条用WHERE order_id IN (...)一次查全部订单项。从20次查询变成1次。
4.2 索引优化：慢查询日志告诉我的
优化后慢查询日志还在刷，但变成了不同的SQL。我看了下执行计划：
EXPLAIN ANALYZE SELECT * FROM order_items
WHERE order_id IN (123, 456, 789, ...) ;
-- 结果：Seq Scan on order_items (cost=0.00..8500.00 rows=800 ...)
order_items 表没有order_id的索引？建表时漏了。加上：
CREATE INDEX CONCURRENTLY idx_order_items_order_id ON order_items (order_id);
CREATE INDEX CONCURRENTLY idx_orders_user_id_created_at ON orders (user_id, created_at DESC);
第二个索引是为了分页排序——ORDER BY created_at DESC LIMIT 20 需要复合索引才能避免排序。
五、缓存策略：从单机LRU到Redis管道
查询优化后，P95从2180ms降到520ms。但还不够，因为数据库压力还是高——每个请求都要查订单表。我需要缓存。
5.1 第一层：进程内LRU缓存（FastAPI+TTL）
from functools import lru_cache
from datetime import datetime, timedelta
@lru_cache(maxsize=1024)
def get_cached_order_ids(user_id: int, page: int, size: int) -> list[int]:
# 只能缓存订单ID，不能缓存整个对象（避免数据一致性问题）
...
但lru_cache不支持TTL，我用了一个技巧：把时间戳编码进key：
from cachetools import TTLCache
order_ids_cache = TTLCache(maxsize=2048, ttl=60) # 60秒过期
async def get_orders_cached(user_id, page, size):
cache_key = f"order_ids:{user_id}:{page}:{size}"
if cache_key in order_ids_cache:
order_ids = order_ids_cache[cache_key]
else:
# 查数据库拿订单ID
stmt = select(Order.id).where(...).limit(size).offset(...)
order_ids = (await session.execute(stmt)).scalars().all()
order_ids_cache[cache_key] = order_ids
# 根据订单ID批量查详情（走Redis缓存）
...
5.2 第二层：Redis管道批量取订单详情
每个订单详情是独立的缓存key，用管道一次性取回：
import redis.asyncio as aioredis
redis_client = aioredis.from_url("redis://...", decode_responses=True)
async def get_order_details_from_cache(order_ids: list[int]) -> dict[int, dict]:
# 使用管道批量获取
pipe = redis_client.pipeline(transaction=False)
for oid in order_ids:
pipe.hgetall(f"order:detail:{oid}") # 用hash存储
results = await pipe.execute()
# 处理缓存未命中的
miss_ids = [oid for oid, data in zip(order_ids, results) if not data]
if miss_ids:
# 批量查数据库
stmt = select(Order).where(Order.id.in_(miss_ids))\
.options(selectinload(Order.items)...)
db_orders = (await session.execute(stmt)).scalars().all()
# 写回缓存
set_pipe = redis_client.pipeline()
for order in db_orders:
order_dict = serialize_order(order)
set_pipe.hset(f"order:detail:{order.id}", mapping=order_dict)
set_pipe.expire(f"order:detail:{order.id}", 300) # 5分钟TTL
await set_pipe.execute()
# 更新返回结果
for order in db_orders:
results[order_ids.index(order.id)] = serialize_order(order)
return {oid: data for oid, data in zip(order_ids, results)}
5.3 踩坑：Redis序列化别用pickle
我用pickle.dumps存过订单对象，结果升级代码后反序列化失败。改用JSON序列化，字段丢失时用default参数兜底：
def serialize_order(order) -> dict:
return {
"id": order.id,
"status": order.status,
"total_amount": str(order.total_amount), # Decimal不能直接JSON序列化
"created_at": order.created_at.isoformat(),
"items": [
{
"product_id": item.product_id,
"name": item.product_snapshot.name,
"price": str(item.price),
"qty": item.quantity,
}
for item in order.items
]
}
六、效果数据：调优前后的对比
压测命令不变，结果如下：
| 指标 | 优化前 | 优化后(查询) | 优化后(查询+缓存) |
|---|---|---|---|
| P50延迟 | 450ms | 120ms | 45ms |
| P95延迟 | 2180ms | 520ms | 90ms |
| P99延迟 | 3400ms | 890ms | 150ms |
| QPS | 430 | 1200 | 2100 |
| 数据库CPU | 95% | 45% | 12% |
| 数据库QPS | 8200 | 3500 | 800 |
内存方面，进程内缓存占用了约120MB（2048个key，每个key存20个订单ID），Redis内存增加了约50MB（缓存了最近的热点订单）。考虑到订单数据本身就不大（每个订单详情约2KB），这个开销可以接受。
七、总结与建议
这次调优的核心思路是：先Profile，再索引，最后缓存。很多人跳过了前两步直接上缓存，结果缓存了N+1查询的结果，只是把慢查询从数据库搬到了Redis——治标不治本。
几个具体的建议：
- SQLAlchemy 2.0中，永远不要用
lazy="select"，要么用selectinload，要么用lazy="raise"强制开发时暴露问题。 - 索引不是越多越好，但复合索引的顺序很重要——
(user_id, created_at DESC)比(created_at DESC)更适合这个场景。 - 缓存一定要分层：进程内缓存应对热点，Redis应对分布式共享。但注意进程内缓存的过期时间要短，否则数据不一致。
- 压测时别只测一个用户，我最初用
user_id=12345压测，缓存命中率100%，数据好看但没意义。换多个用户ID后，缓存命中率降到70%，这才是真实情况。
最后留个问题：如果订单量继续增长，Redis缓存占满内存怎么办？我的方案是只缓存最近7天的订单，更早的订单直接查数据库。这个策略我们在下一个版本上线，到时候再写一篇分享。
