1. 问题背景：一个本不该慢的接口
上周五，运维同事在群里甩了个截图：/api/v1/orders 接口在高峰期P95耗时飙到1.2s，QPS只有可怜的230。客户投诉说订单列表加载像“幻灯片”。
我第一反应是“这接口逻辑不复杂啊，就查个订单表关联用户和商品，加个分页”。但线上数据不会撒谎。排查的第一步，不是看代码，而是先复现和测量。
环境信息：
- Python 3.10.12
- FastAPI 0.104.1（对比Flask 2.3.3）
- SQLAlchemy 2.0.23 + PostgreSQL 15.3
- Redis 7.0.12
- Gunicorn 21.2.0（4 workers, sync worker class）
- 压测工具：wrk 4.2.0
2. Profiling：别猜，用数据说话
我习惯先用cProfile跑一次单请求，看函数级耗时分布。但cProfile对异步代码支持一般，所以FastAPI应用我用Py-Spy（采样式，不阻塞业务）：
# 安装
pip install py-spy
# 对运行中的FastAPI进程做10秒采样
sudo py-spy record --pid $(pgrep -f uvicorn) -o profile.svg --duration 10
生成的火焰图让我一眼看到问题：SQLAlchemy的_fetchall_with_flat_async占了63%的时间。再细看，是ORM懒加载导致的N+1查询——查了100个订单，每个订单又单独查用户和商品，共201条SQL。
Flask侧（同步版）用cProfile更直观：
import cProfile
import pstats
from app import app
def run_request():
client = app.test_client()
client.get('/api/v1/orders?page=1&size=50')
cProfile.run('run_request()', 'prof.out')
p = pstats.Stats('prof.out')
p.sort_stats('cumulative').print_stats(15)
输出明确显示：sqlalchemy.orm.loading._emit_load 累计耗时0.42s，占了总时间78%。
结论：瓶颈不在框架本身，而在数据库访问模式。
3. 数据库查询优化：干掉N+1
原代码（典型的懒加载陷阱）：
# 问题代码：orders查询后，访问order.user和order.items才触发懒加载
async def get_orders(db: AsyncSession, page: int, size: int):
stmt = select(Order).offset((page-1)*size).limit(size)
result = await db.execute(stmt)
orders = result.scalars().all()
# 下面这行会触发2*N条额外SQL
return [serialize_order(o) for o in orders] # 内部访问o.user, o.items
重构为显式join + selectinload（一次SQL联合查，或分两次批量IN查询）：
from sqlalchemy.orm import selectinload
from sqlalchemy import select
async def get_orders_optimized(db: AsyncSession, page: int, size: int):
stmt = (
select(Order)
.options(
selectinload(Order.user), # 生成 IN 子查询批量加载
selectinload(Order.items).selectinload(Item.product)
)
.order_by(Order.created_at.desc())
.offset((page-1)*size)
.limit(size)
)
result = await db.execute(stmt)
orders = result.scalars().unique().all() # 注意：unique()必须加，防止行重复
return [serialize_order(o) for o in orders]
同时给orders.user_id和orders.created_at加了复合索引：
CREATE INDEX idx_orders_user_created ON orders (user_id, created_at DESC);
效果对比（wrk压测，60秒，100并发）：
| 指标 | 优化前 | 优化后 | 提升 |
|---|---|---|---|
| SQL执行数 | 201 | 3 | 98.5% |
| 平均耗时 | 180ms | 74ms | 59% |
| QPS | 320 | 780 | 144% |
数据库查询优化立竿见影，但还不够。74ms延迟对高并发场景还是偏高。
4. 缓存策略：Redis + 主动失效
订单列表是读多写少（读写比约15:1），非常适合加缓存。我的设计：
- Cache-Aside模式：先读Redis，miss则查DB并回填。
- TTL设为60秒，防止数据长时间脏读。
- 主动失效：下单、修改订单状态时，删除对应用户的列表缓存。
import json
import redis.asyncio as aioredis
redis_client = aioredis.from_url("redis://localhost:6379/0", decode_responses=True)
async def get_orders_cached(db, user_id, page, size):
cache_key = f"orders:user:{user_id}:page:{page}:size:{size}"
# 1. 读缓存
cached = await redis_client.get(cache_key)
if cached:
return json.loads(cached)
# 2. 缓存miss，查DB（用上面优化后的查询）
orders = await get_orders_optimized(db, page, size)
# 3. 回填缓存，TTL 60s
await redis_client.set(
cache_key,
json.dumps(orders, default=str),
ex=60
)
return orders
# 写操作时主动失效
async def create_order(db, user_id, order_data):
# ... 业务逻辑 ...
# 删除该用户所有页的缓存（用scan批量删）
async for key in redis_client.scan_iter(f"orders:user:{user_id}:*"):
await redis_client.delete(key)
注意： 别用flushall，生产环境会出事故。我用scan_iter批量删，避免阻塞Redis单线程。
加了缓存的压测结果：
| 指标 | 查询优化后 | +Redis缓存 |
|---|---|---|
| 平均耗时 | 74ms | 23ms |
| P95 | 112ms | 38ms |
| QPS | 780 | 2100 |
| DB QPS | 780 | 42（缓存命中率95%） |
缓存命中率我用Redis的INFO stats命令确认：keyspace_hits:keyspace_misses = 15230:801，约95%。
5. 踩坑与细节：FastAPI vs Flask的隐藏差异
调优过程中踩了几个坑，值得记录：
坑1：Flask同步框架的线程阻塞
Flask + Gunicorn sync worker下，每个请求占一个线程，长DB查询会耗尽线程池。我对比了Flask版本（同样优化+缓存后），QPS只能到1450，比FastAPI低31%。原因在于FastAPI的异步SQLAlchemy能单线程处理更多并发请求。如果项目必须用Flask，建议换gevent worker：
gunicorn -w 4 -k gevent --worker-connections 1000 app:app
但gevent对psycopg2需要补丁，得用psycogreen，麻烦。
坑2：selectinload导致的内存膨胀
如果订单列表要返回几千条，selectinload会把所有关联对象一次性载入内存。我在压测200并发时遇到内存从300MB涨到1.2GB。解决方案：**强制分页固定`size 100:
raise HTTPException(400, "size max is 100")
**坑3：Redis连接池耗尽**
刚开始用`redis.asyncio`每次请求新建连接，压测时报`ConnectionPoolError`。改为全局单例连接池：
```python
redis_client = aioredis.from_url(
"redis://localhost:6379/0",
max_connections=50,
decode_responses=True
)
坑4：缓存序列化性能
用json.dumps序列化ORM对象很慢（占缓存路径40%耗时）。改用pickle.dumps（快1.8倍），但要注意pickle安全性——只缓存内部数据，不暴露给客户端。
6. 最终效果与总结
完整优化链路后的最终数据（wrk -t8 -c100 -d60s）：
| 指标 | 初始状态 | 最终状态 | 提升 |
|---|---|---|---|
| 平均延迟 | 180ms | 23ms | 87.2% |
| P95延迟 | 420ms | 38ms | 91.0% |
| QPS | 320 | 2100 | 556% |
| CPU使用率 | 85% | 48% | -43% |
调优优先级结论：
1. 先profiling，不然后续优化全是盲枪——Py-Spy火焰图20分钟定位问题。
2. 数据库优化收益最大（N+1修复带来144%QPS提升），且是缓存之前的基础。
3. 缓存是倍增器，但前提是查询本身不能烂——给烂查询加缓存只是把垃圾缓存住。
4. 框架选择：异步框架（FastAPI）在高并发IO密集型场景下比Flask（同步）有明确优势，但如果项目已用Flask，gevent + 合理索引也能救。
最后说句掏心窝的：性能调优最怕“我觉得”。每次改动，都跑一轮压测，记录数据，用事实说话。上面所有数据来自同一台测试机（8核16G），生产环境建议用gunicorn --preload预加载模型，避免worker内存翻倍。
有疑问欢迎留言，踩坑不易，互相救火。
（本文所有压测数据可复现，代码基于Python 3.10，SQLAlchemy 2.x，Redis 7.x，若有版本差异请调整对应API。）
