一、问题背景：一个“看起来没问题”的接口
上个月接了个活，给一个电商后台的订单列表接口做性能优化。这个接口逻辑很简单：根据用户ID分页查订单，返回订单及商品快照。上线前压测发现P95延迟高达1.2秒，QPS只有85，数据库CPU直接飙到80%。
第一反应是“查一下慢SQL”。结果EXPLAIN一看，三个表（orders、order_items、products）都是全表扫描，单次查询耗时380ms。但更让人意外的是，真正的问题不在SQL本身，而在应用层的N+1查询——每查一页20条订单，就要额外执行20次商品快照查询。
二、环境与版本：统一基线
调优前先固定环境，避免变量干扰：
- Python 3.10.12（GIL的影响后面会提到）
- FastAPI 0.104.0 + Uvicorn 0.24.0（worker=4, loop=uvloop）
- Flask 2.3.3 + Gunicorn 21.2.4（worker=4, 同步worker + gevent）
- SQLAlchemy 2.0.21（async for FastAPI，sync for Flask）
- PostgreSQL 15.2（shared_buffers=1GB, work_mem=32MB）
- Redis 7.0（maxmemory-policy=allkeys-lru）
- 压测工具：wrk 2.0（12线程，100连接，持续60秒）
压测脚本统一用：
wrk -t12 -c100 -d60s --latency http://localhost:8000/api/orders?user_id=123&page=1&size=20
三、第一步：Profiling定位瓶颈——cProfile + py-spy
别猜，直接测。我先用cProfile跑了一次单请求：
python -m cProfile -s cumulative my_api.py
结果前几行输出：
ncalls tottime percall cumtime percall filename:lineno(function)
1 0.002 0.002 0.583 0.583 /app/api/orders.py:22(get_orders)
20 0.001 0.000 0.412 0.020 /app/models.py:145(get_product_snapshot)
20 0.003 0.000 0.310 0.015 /app/serializer.py:88(order_item_to_dict)
一眼看出两个大坑：
- 20次
get_product_snapshot累计0.412秒——典型的N+1查询。 - 序列化
order_item_to_dict累计0.310秒——每调一次就做一次Decimal到float转换、datetime格式化，20个订单项*5个商品快照=100次序列化调用。
这时候用py-spy抓一下线上运行的火焰图（因为cProfile在异步环境下不准）：
py-spy record -o flame.svg --pid --duration 30
火焰图确认了：SQLAlchemy session.get()占据了58%的CPU时间，jsonable_encoder占22%。
结论：先别碰缓存，把查询和序列化优化了再说。
四、第二步：数据库优化——Eager Loading + 复合索引
4.1 修复N+1查询
原代码（伪代码，但和真实逻辑一致）：
# 优化前：每页20条订单，每条订单查一次商品快照
async def get_orders(user_id, page, size):
orders = await db.execute(
select(Order).where(Order.user_id == user_id)
.offset(page * size).limit(size)
)
result = []
for order in orders.scalars().all():
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)
result.append(serialize(order, item, product))
return result
优化后：用SQLAlchemy 2.0的selectinload一次性加载关联对象。
# 优化后：一次join拉取所有关联数据
async def get_orders(db: AsyncSession, user_id: int, page: int, size: int):
stmt = (
select(Order)
.options(
selectinload(Order.items).selectinload(OrderItem.product) # 关键：级联预加载
)
.where(Order.user_id == user_id)
.order_by(Order.created_at.desc())
.offset(page * size)
.limit(size)
)
result = await db.execute(stmt)
orders = result.scalars().unique().all() # 注意：must call unique() for selectinload
return [serialize_order(o) for o in orders]
4.2 加复合索引
原表索引只有主键。通过pg_stat_statements找到高频查询，添加：
CREATE INDEX idx_orders_user_created ON orders (user_id, created_at DESC) INCLUDE (status, total_amount);
CREATE INDEX idx_order_items_order_id ON order_items (order_id) INCLUDE (product_id, quantity, price);
CREATE INDEX idx_products_id ON products (id) INCLUDE (name, image_url, sku);
覆盖索引让查询只需要走索引扫描，不需要回表。
优化后单次查询耗时从380ms降到41ms，N+1彻底消失。
五、第三步：序列化优化——手写Pydantic validator
FastAPI默认用jsonable_encoder，慢且耗CPU。我的做法：给Pydantic模型加@field_serializer，让Decimal和datetime在模型层直接转好，避免二次处理。
from pydantic import BaseModel, field_serializer
from datetime import datetime
from decimal import Decimal
class OrderItemOut(BaseModel):
product_name: str
quantity: int
unit_price: Decimal
total_price: Decimal
created_at: datetime
@field_serializer('unit_price', 'total_price')
def serialize_decimal(self, value: Decimal) -> float:
return float(value) # 保留两位小数，避免精度丢失
@field_serializer('created_at')
def serialize_dt(self, value: datetime) -> str:
return value.strftime('%Y-%m-%d %H:%M:%S')
同时把response_model=OrderOut显式声明，FastAPI就不会走jsonable_encoder，而是直接用Pydantic的model_dump_json——速度提升约70%。
序列化这一步，从310ms降到75ms。
六、第四步：缓存策略——Redis缓存热点用户
数据库优化后，单次查询约120ms。但压测发现QPS到不了300——因为PostgreSQL的CPU还是高（约40%）。这时候上缓存。
缓存策略：只缓存活跃用户的前3页，避免冷数据污染。TTL设5分钟，因为订单状态变更不太频繁。
# 缓存装饰器实现（FastAPI版本）
from redis.asyncio import Redis
from fastapi import Depends
import json
redis_client = Redis(host='localhost', port=6379, decode_responses=True)
async def cached_orders(user_id: int, page: int, size: int):
cache_key = f"orders:{user_id}:{page}:{size}"
cached = await redis_client.get(cache_key)
if cached:
return json.loads(cached)
# 未命中，查询数据库并写入缓存
orders = await fetch_orders_from_db(user_id, page, size)
await redis_client.setex(cache_key, 300, json.dumps(orders))
return orders
缓存命中率在压测场景下约78%（因为wrk的user_id从1到1000均匀分布，前200个用户占了80%的请求）。命中时响应时间6ms，未命中时120ms，整体平均降到38ms。
七、踩坑与优化：三个隐藏的坑
- SQLAlchemy 2.0的
unique()陷阱：用了selectinload后，如果结果集有多对多关系，必须调用.unique()，否则返回重复对象。少写这个，压测时数据全是错的，白调了两天。 - Redis连接池：FastAPI的异步Redis默认连接池上限50，压测时并发100，直接爆连接错误。要手动设置
max_connections=200，否则缓存反而成了瓶颈。 - GIL对Flask的影响：Flask用同步worker时，GIL导致CPU密集的序列化操作无法并行。换成gevent后好了点，但整体还是比FastAPI差一截——FastAPI的异步IO + uvicorn的uvloop在处理IO密集场景下优势明显。
八、压测数据：FastAPI vs Flask
统一优化后（Eager Loading + 索引 + Redis），wrk压测结果对比：
| 指标 | FastAPI (uvicorn 4w) | Flask (gunicorn+gevent 4w) |
|---|---|---|
| 平均延迟 | 87ms | 134ms |
| P95延迟 | 152ms | 245ms |
| QPS | 420 | 285 |
| 数据库CPU | 25% | 38% |
| Redis命中率 | 78% | 78% |
调优前FastAPI是1200ms/85QPS，调优后87ms/420QPS，整体提升约5倍。
Flask的瓶颈在GIL——即使加了gevent，序列化还是串行的。如果生产环境必须用Flask，建议把序列化逻辑改成orjson或者用multiprocessing做CPU隔离。
九、总结与建议
调优的优先级是：先profiling，再查SQL，最后上缓存。别一上来就套Redis，很可能你的数据库查询还有10倍优化空间。
核心收获：
- cProfile/py-spy定位N+1查询和序列化热点
- SQLAlchemy selectinload + 覆盖索引，查询从380ms降到41ms
- Pydantic @field_serializer 替代 jsonable_encoder，序列化从310ms降到75ms
- Redis缓存热点用户前3页，TTL 300s，命中率78%
如果你也在用FastAPI或Flask，建议先跑一下py-spy dump --pid看看你的CPU到底花在哪了——很可能不是数据库，而是你的代码自己。
最后说一句：没有银弹，只有一步步压出来的数据。
