一、问题背景：从监控告警到性能危机
上周四下午，运维连续弹出3条告警：order/detail接口P95延迟超过3s，错误率飙至7%。这个接口是B端订单列表页的核心依赖，每次页面加载要拉取50条订单及关联的商品、用户、物流信息。起初以为是大促流量冲击，但看了监控发现QPS只有200——这个量级对FastAPI来说本该是“洒洒水”。
直觉告诉我，问题不在框架层，而在业务代码的查询方式上。果然，用py-spy dump线程栈，发现大量线程卡在SQLAlchemy的lazy load上——典型的N+1查询。但N+1只是冰山一角，更深层的问题是：每次请求都在重复计算订单的“优惠分摊金额”，这个计算涉及3张表的聚合，耗时约200ms。
二、环境与版本：一切性能讨论的前提
- Python 3.11.4（GIL影响显著，异步IO收益大）
- FastAPI 0.104.0 + Uvicorn 0.24.0（单worker模式，方便对比）
- SQLAlchemy 2.0.23（async模式）+ asyncpg 0.28.0
- Redis 7.0（单实例，无集群）
- 压测工具：wrk 4.2.0（macOS本机，关闭HTTP keepalive）
- 服务部署：4C8G Docker容器，PostgreSQL 15（默认配置，未调shared_buffers）
所有测试均在同一容器内执行，排除网络抖动因素。
三、第一步：用cProfile定位瓶颈，别靠猜
很多同学遇到性能问题第一反应是加缓存，这其实是偷懒。先profiling，用数据说话。我写了个最小化调用脚本，直接在本地触发接口逻辑：
# perf_profiler.py
import cProfile
import pstats
import asyncio
from app.services.order import get_order_detail
async def run():
# 模拟真实请求参数：50条订单
await get_order_detail(user_id=12345, page=1, page_size=50)
if __name__ == "__main__":
profiler = cProfile.Profile()
profiler.enable()
asyncio.run(run())
profiler.disable()
stats = pstats.Stats(profiler)
stats.sort_stats('cumtime') # 按累计时间排序
stats.print_stats(30) # 打印前30行
运行后关键输出（已脱敏）：
ncalls tottime cumtime filename:lineno
1 0.002 1.902 order.py:112(get_order_detail)
50 0.001 1.150 order.py:156(_load_items) # 商品查询
150 0.003 0.980 sqlalchemy/orm/loading.py:264 # lazy load
50 0.002 0.452 order.py:203(_calc_discount) # 优惠计算
50 0.001 0.301 order.py:220(_query_coupon) # 优惠券查询
分析结论：
- _load_items累计耗时1.15s，里面包含50次独立的商品查询（N+1）
- _calc_discount每次耗时9ms，但调用了50次，且内部还有子查询
- 总耗时1.9s，和压测的P95吻合
这里有个小坑：cProfile在异步代码中会显示
await点，但不会展示IO等待时间。所以我又用了py-spy dump --pid看真实阻塞，确认是数据库IO等待。
四、数据库查询优化：selectinload与聚合下推
方案设计
- N+1修复：将
lazy="select"改为lazy="selectin"，一次JOIN批量加载商品和用户信息 - 优惠计算下推：把Python中的循环计算改为SQL聚合（
SUM+GROUP BY），减少数据传输 - 索引补充：发现
order_item表缺少(order_id)联合索引，补一个
核心实现
# app/repositories/order_repo.py
from sqlalchemy import select, func
from sqlalchemy.orm import selectinload
async def get_order_list_with_items(session, user_id: int, page: int, size: int):
# 修复前：lazy load导致50次SQL
# stmt = select(Order).where(Order.user_id == user_id)
# 修复后：一次JOIN加载所有关联对象
stmt = (
select(Order)
.options(
selectinload(Order.items), # 商品列表
selectinload(Order.user), # 用户信息
selectinload(Order.logistics), # 物流信息
)
.where(Order.user_id == user_id)
.order_by(Order.created_at.desc())
.limit(size)
.offset((page - 1) * size)
)
result = await session.execute(stmt)
return result.scalars().unique().all()
# 优惠分摊计算：从Python循环改为SQL聚合
async def get_discount_summary(session, order_ids: list[int]):
stmt = (
select(
OrderItem.order_id,
func.sum(OrderItem.discount_amount).label("total_discount"),
func.count(OrderItem.id).label("item_count"),
)
.where(OrderItem.order_id.in_(order_ids))
.group_by(OrderItem.order_id)
)
result = await session.execute(stmt)
return {row.order_id: row for row in result}
踩坑记录
- selectinload与分页冲突：如果先
limit再selectinload，SQLAlchemy会先查主表再查关联表（两条SQL），没问题。但如果你用了distinct，会导致关联表数据翻倍——因为JOIN后再DISTINCT会去重主表行，但关联表行数变了。 - 索引命名：
CREATE INDEX CONCURRENTLY idx_order_item_order_id ON order_item(order_id)，这里务必用CONCURRENTLY，否则锁表。
效果数据
| 指标 | 优化前 | 优化后 | 提升幅度 |
|---|---|---|---|
| SQL查询次数 | 152次 | 4次 | 97.4% |
| 接口耗时 | 1900ms | 480ms | 74.7% |
| P95延迟 | 3.8s | 720ms | 81.1% |
数据库CPU从85%降至30%，这步立竿见影。
五、缓存策略：本地缓存+Redis两级缓存
数据库优化后耗时480ms，但QPS 200时CPU已接近60%。瓶颈转移到了Python对象的序列化与数据库连接池。此时才轮到缓存出场——缓存是最后的手段，不是第一手段。
方案设计
- 一级缓存：进程内
functools.lru_cache，TTL 5秒，适合热点数据（如商品名称、用户昵称） - 二级缓存：Redis，TTL 60秒，存储订单详情JSON
- 失效策略：写操作时主动删除Redis key（不更新，防止并发写旧值）
核心实现
# app/cache.py
import json
import redis.asyncio as aioredis
from functools import lru_cache
redis_client = aioredis.from_url(
"redis://localhost:6379/0",
max_connections=50,
decode_responses=True
)
def order_cache_key(order_id: int) -> str:
return f"order:detail:{order_id}"
async def get_cached_order(order_id: int):
# 一级缓存：本地进程缓存（5秒）
local_key = f"order:{order_id}"
local_data = local_cache.get(local_key)
if local_data:
return local_data
# 二级缓存：Redis（60秒）
data = await redis_client.get(order_cache_key(order_id))
if data:
# 回填本地缓存
local_cache.set(local_key, json.loads(data), ttl=5)
return json.loads(data)
return None
async def set_order_cache(order_id: int, data: dict):
await redis_client.setex(
order_cache_key(order_id),
60, # TTL 60秒
json.dumps(data, ensure_ascii=False)
)
# lru_cache装饰器示例：用于不常变化的数据
@lru_cache(maxsize=256, ttl=5) # 注意：官方lru_cache不支持ttl，这里是伪代码
def get_product_name(product_id: int):
# 实际项目中我会用cachetools.TTLCache
return db_query(...)
实际使用cachetools库，它支持TTL：
from cachetools import TTLCache
product_cache = TTLCache(maxsize=1024, ttl=5)
def get_product_name_cached(product_id: int):
if product_id in product_cache:
return product_cache[product_id]
name = db_query(...)
product_cache[product_id] = name
return name
踩坑与优化
- Redis序列化：用
pickle比json快30%，但跨版本不安全。建议用msgpack，体积小且快。 - 缓存穿透：压测时发现大量请求查不存在的订单，直接打到DB。加了布隆过滤器，代码省略，思路是启动时加载所有有效订单ID。
- 缓存雪崩：TTL加随机偏移量，避免同一时刻大面积失效。这里用
ttl = 60 + random.randint(0, 10)。
效果数据
| 指标 | 数据库优化后 | 加缓存后 | 提升幅度 |
|---|---|---|---|
| 接口耗时 | 480ms | 180ms | 62.5% |
| QPS | 350 | 1200 | 242.9% |
| Redis内存 | - | 85MB | - |
六、最终压测对比：wrk数据说话
压测命令（保持连接30秒，8线程，200连接）：
wrk -t8 -c200 -d30s --latency http://localhost:8000/order/detail?user_id=12345&page=1
优化前（基线）：
Requests/sec: 204.85
Latency Distribution
50% 1.92s
75% 2.71s
90% 3.45s
99% 4.02s
优化后（全套）：
Requests/sec: 1843.62
Latency Distribution
50% 142ms
75% 188ms
90% 245ms
99% 380ms
总结：
- QPS从205提升至1843，提升约9倍
- P99从4.02s降至380ms，提升10.5倍
- 整个过程：profiling占30%时间，SQL优化占40%，缓存设计占30%
最后强调两点：
1. 先压测再优化：不压测就不知道瓶颈在哪，我见过太多团队盲目加缓存，结果数据库连接池先爆了。
2. 缓存是手段不是目的：如果数据库查询本身烂，缓存只会掩盖问题。我这次先把SQL改成selectinload+聚合，才敢上缓存。
代码已上传至GitHub（搜fastapi-order-performance），有问题评论区聊。
