一、问题背景：一个“正常”接口的突然变慢
上周三下午，运维老张甩给我一张监控截图：订单列表接口 /api/orders?user_id=1024 在晚高峰时段P95延迟从600ms飙升至2.1s，QPS从380掉到220。我看了一眼代码——FastAPI + SQLAlchemy 2.0 + PostgreSQL 14，逻辑很简单：查用户订单，关联商品信息，再查一次卖家店铺名。直觉告诉我，问题出在ORM的懒加载上。
但作为老开发，我坚持“先测量再优化”。于是搬出三件套：cProfile（函数级耗时）、py-spy（线上进程无侵入采样）、pg_stat_statements（数据库层慢查询）。如果你还在用 time.time() 手动打点，建议直接跳到第三节。
二、环境与版本：统一压测基线
为了公平对比，我同时跑了Flask（2.3.2）和FastAPI（0.95.1）两个服务，共用同一套ORM模型和数据库。
# 压测环境
Python: 3.11.4
FastAPI: 0.95.1 + uvicorn 0.23.2
Flask: 2.3.2 + gunicorn 21.2.0 (gevent worker)
SQLAlchemy: 2.0.19
PostgreSQL: 14.8 (shared_buffers=4GB, effective_cache_size=12GB)
Redis: 7.0.12 (maxmemory 2GB, allkeys-lru)
压测工具: wrk 4.2.0 (线程8, 连接400, 时长60s)
启动命令注意：FastAPI用 uvicorn main:app --workers 4 --loop uvloop，Flask用 gunicorn -w 4 -k gevent --worker-connections 1000 wsgi:app。别用Flask默认的同步worker，会死得很惨。
三、Profiling第一步：找到罪魁祸首
先用py-spy对运行中的FastAPI进程采样10秒：
py-spy dump --pid 31245 --duration 10 > fastapi_profile.txt
输出片段揭示了真相：
Thread 0x7f8a1c (idle):
File "sqlalchemy/orm/loading.py:112" in _instance_processor
File "sqlalchemy/orm/query.py:1280" in _iter_expr
File "app/order_service.py:45" in get_orders
- 87.2% of total time
87%的时间在ORM加载实例！再看 pg_stat_statements 输出，发现一个奇怪现象：同样的SQL SELECT * FROM orders WHERE user_id=$1 LIMIT 20 执行了23次，但只有前两次是真正的数据库查询，其余是Python层的懒加载触发。
结论： 典型的N+1查询。SQLAlchemy 2.0默认是懒加载，我在查询订单后访问 order.items 和 order.seller.name，每次访问都触发一次新的SQL。20条订单 = 1次主查询 + 20次商品查询 + 20次店铺查询 = 41次往返。
四、方案设计：从三个层面干掉瓶颈
第一刀：ORM查询优化（消除N+1）
用 joinedload() 或 selectinload() 显式预加载。这里有个坑：joinedload 对集合类型（list）会生成笛卡尔积，如果订单有大量商品项，结果集膨胀10倍。所以我选 selectinload，它生成第二条IN查询，避免笛卡尔积。
第二刀：数据库索引与查询重写
检查表结构，发现 orders.user_id 有索引，但 orders.status 没有。优化后的查询加了 status='paid' 过滤，必须加复合索引。另外，order_items 表外键 order_id 缺索引，导致商品关联查询走Seq Scan。
第三刀：Redis缓存（读多写少场景）
订单列表属于读多写少（读写比约8:2），且用户不会立即感知数据变化。引入Redis缓存，TTL设90秒，用 user_id:page 作为key。特别注意：缓存击穿（突然过期）用互斥锁解决，缓存雪崩加随机过期时间。
五、核心实现：代码与踩坑实录
1. 修复N+1查询（FastAPI + SQLAlchemy）
# app/order_service.py 优化后
from sqlalchemy.orm import selectinload, Session
from app.models import Order, OrderItem, Seller
def get_orders(db: Session, user_id: int, page: int = 1):
query = (
db.query(Order)
.filter(Order.user_id == user_id, Order.status == 'paid')
.options(
selectinload(Order.items).joinedload(OrderItem.product),
selectinload(Order.seller).joinedload(Seller.shop_name) # 注意嵌套
)
.order_by(Order.created_at.desc())
.offset((page - 1) * 20)
.limit(20)
)
return query.all()
踩坑1： 我用 joinedload(Order.items).joinedload(OrderItem.product) 时，SQLAlchemy 2.0直接报错 AttributeError: 'Load' object has no attribute 'items'。原因是 items 是集合，必须用 selectinload(Order.items).joinedload(OrderItem.product) 这种链式组合。集合用selectinload，标量用joinedload。
踩坑2： 加了 status='paid' 过滤后，原来跑得飞快的查询突然变慢。检查执行计划发现PostgreSQL没用上 user_id 索引，因为复合条件导致选择性降低。解决方案：创建 (user_id, status, created_at) 复合索引，并ANALYZE更新统计信息。
CREATE INDEX idx_orders_user_status_created ON orders (user_id, status, created_at DESC);
CREATE INDEX idx_order_items_order_id ON order_items (order_id);
ANALYZE orders;
2. Redis缓存装饰器（通用方案）
# app/cache.py
import json
import hashlib
import redis
from functools import wraps
from fastapi import HTTPException
r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
def cache_response(ttl: int = 90, prefix: str = "api"):
def decorator(func):
@wraps(func)
async def wrapper(*args, **kwargs):
# 将参数序列化为key的一部分
key_data = json.dumps({
"args": args[1:], # 跳过self
"kwargs": kwargs,
}, sort_keys=True)
cache_key = f"{prefix}:{func.__name__}:{hashlib.md5(key_data.encode()).hexdigest()}"
# 尝试读缓存
cached = r.get(cache_key)
if cached is not None:
return json.loads(cached)
# 互斥锁防止缓存击穿
lock_key = f"lock:{cache_key}"
if r.set(lock_key, "1", nx=True, ex=5):
try:
result = await func(*args, **kwargs)
# 随机ttl防雪崩
actual_ttl = ttl + random.randint(0, 30)
r.setex(cache_key, actual_ttl, json.dumps(result, default=str))
return result
finally:
r.delete(lock_key)
else:
# 获取到锁的线程正在写缓存，等待重试（此处简化处理）
raise HTTPException(status_code=503, detail="Try again later")
return wrapper
return decorator
# 使用示例
@cache_response(ttl=90, prefix="order")
async def get_order_list(user_id: int, page: int = 1):
# 原本的数据库查询逻辑
pass
踩坑3： 异步函数里不能直接用 time.sleep 做重试，必须 await asyncio.sleep。且FastAPI的依赖注入中 Request 对象不能序列化，所以key生成时过滤掉了非基础类型参数。
3. Flask侧同样的优化（附对比）
Flask代码大同小异，但要注意：Flask默认同步视图，如果用了 selectinload，线程会阻塞在IO上。建议用 gevent worker，并在 __init__.py 里打补丁：
from gevent import monkey
monkey.patch_all()
否则你会看到SQLAlchemy连接池被占满，报 QueuePool limit of size 5 overflow 10 reached。
六、压测数据：优化前后对比
我用wrk跑三轮，取中位数。测试命令：
wrk -t8 -c400 -d60s --latency http://localhost:8000/api/orders?user_id=1024
优化前（FastAPI + 懒加载）：
| 指标 | 数值 |
|---|---|
| QPS | 220 req/s |
| P50 | 380ms |
| P95 | 2.1s |
| P99 | 4.8s |
| 错误率 | 0.5%（超时） |
| 数据库查询次数/请求 | 41 |
优化后（FastAPI + selectinload + Redis缓存）：
| 指标 | 数值 |
|---|---|
| QPS | 1450 req/s |
| P50 | 45ms |
| P95 | 180ms |
| P99 | 250ms |
| 错误率 | 0% |
| 数据库查询次数/请求 | 2（缓存命中时0） |
Flask优化后：
| 指标 | 数值 |
|---|---|
| QPS | 980 req/s |
| P50 | 72ms |
| P95 | 300ms |
Flask比FastAPI慢约35%，差距主要在异步框架的协程调度上。但如果只做同步IO操作，gevent worker能把差距缩小到20%以内。结论：如果你的代码全是最优的，FastAPI的异步优势在纯数据库操作下不明显；但一旦有多路IO（如调外部API+读DB），差距会拉大到3倍以上。
七、总结与反思
- 不要相信直觉：我最初怀疑是JSON序列化慢，结果profiling显示是ORM。cProfile和py-spy能救你于水火。
- 数据库是第一战场：N+1查询是万恶之源，写ORM前先想清楚关联深度。用
echo=True或slowquery_log把SQL打印出来。 - 缓存是放大器：加了Redis后QPS提升6倍，但要注意缓存一致性。我的场景允许90秒延迟同步，如果你做支付系统，别这么干。
- 框架选型：FastAPI的异步是锦上添花，不是雪中送炭。但它的自动OpenAPI文档和类型提示，开发效率确实比Flask高。
最后留个问题：如果订单量过亿，分页到第1000页时，OFFSET 20000 会拖慢查询。你会怎么优化？欢迎评论区讨论。
（附：完整的压测脚本和profiling工具链配置已上传至GitHub，链接见评论区。）
