1. 问题背景：一个“看起来还行”的接口
项目是某内部数据平台的订单查询服务，技术栈FastAPI + SQLAlchemy 2.0 + PostgreSQL 15。某日运维反馈，/api/v1/orders?status=PAID&page=2 这个接口在高峰期P95延迟超过300ms，且数据库CPU飙到70%。
我第一反应是“加个索引不就行了”，但讲真，这种话我自己都不信。先用 time curl 测了一下：
$ curl -o /dev/null -s -w 'total: %{time_total}s\n' 'http://localhost:8000/api/v1/orders?status=PAID&page=2'
total: 0.312s
单次请求312ms，对于内部系统来说不算灾难，但看监控曲线，这个接口每天调用量在200万次以上，P95持续恶化。这不是偶发慢查询，是系统性的性能缺陷。
环境版本（别嫌啰嗦，版本差异真的会坑人）：
- Python 3.11.6 / FastAPI 0.104.1 / Uvicorn 0.24.0
- SQLAlchemy 2.0.23（注意是2.x，ORM写法跟1.x差别很大）
- PostgreSQL 15.3 / Redis 7.2.3
- 压测工具：wrk 4.2.0 / py-spy 0.3.14
2. 第一板斧：Profiling——别猜，直接看火焰图
很多同学调优全靠猜，我过去也这样。后来学乖了：先用工具定位，再动手改代码。这里我用了两个工具，互补：
2.1 cProfile：函数级CPU耗时
写个脚本直接调接口函数，不走网络IO：
# profile_api.py
import cProfile, pstats, io
from fastapi.testclient import TestClient
from main import app
client = TestClient(app)
pr = cProfile.Profile()
pr.enable()
# 模拟真实请求参数
for _ in range(100):
client.get("/api/v1/orders?status=PAID&page=2&page_size=50")
pr.disable()
s = io.StringIO()
ps = pstats.Stats(pr, stream=s).sort_stats("cumulative")
ps.print_stats(20)
print(s.getvalue())
关键输出（截取前10行）：
ncalls tottime percall cumtime percall filename:lineno
100 0.002 0.000 0.289 0.003 orders.py:45(get_orders)
100 0.001 0.000 0.241 0.002 orders.py:78(_serialize_orders)
100 0.011 0.000 0.198 0.002 sqlalchemy/orm/loading.py:128(load)
100 0.002 0.000 0.142 0.001 sqlalchemy/orm/strategies.py:210(selectinload)
5000 0.089 0.000 0.089 0.000 sqlalchemy/orm/attributes.py:600(get_history)
看到没？序列化花了241ms，ORM加载花了198ms，而SQL执行本身只有几十ms。这彻底颠覆了我的直觉——我以为是SQL慢，结果是Python层面的对象转换和N+1查询在作祟。
2.2 py-spy：采样线上进程
cProfile只能测本地，线上问题用py-spy抓实时栈：
# 找到workers的PID
$ ps aux | grep uvicorn
$ sudo py-spy dump --pid 12345
抓到的栈显示，大量线程阻塞在jsonable_encoder和BaseModel.dict()上。这印证了cProfile的结论：瓶颈在序列化。
3. 第二板斧：数据库查询优化——N+1是万恶之源
3.1 复现N+1查询
看代码，问题一目了然。订单表和订单明细表是一对多关系，但我用了懒加载：
# 优化前：懒加载导致N+1查询
class Order(Base):
__tablename__ = "orders"
id = Column(Integer, primary_key=True)
status = Column(String(20), index=True)
items = relationship("OrderItem", lazy="select") # 默认懒加载
# 业务代码
def get_orders(status: str, page: int, page_size: int):
stmt = select(Order).where(Order.status == status)
result = db.execute(stmt).scalars().all()
# 每访问一次 order.items 就发一条SQL
return [{"id": o.id, "items": [i.name for i in o.items]} for o in result]
每页50条订单，就需要50次额外的SELECT * FROM order_items WHERE order_id = ?。加上SQLAlchemy的identity map开销，总查询量是51次！
用echo=True看SQL日志，刷屏刷得怀疑人生。
3.2 优化方案：selectinload + 联合索引
两处改动：
第一处，ORM加载策略改为selectinload（一次IN查询替代N次单查）：
# 优化后：显式批量加载
from sqlalchemy.orm import selectinload
def get_orders(status: str, page: int, page_size: int):
stmt = (
select(Order)
.where(Order.status == status)
.options(selectinload(Order.items)) # 关键：一次IN查询
.order_by(Order.created_at.desc())
.offset((page - 1) * page_size)
.limit(page_size)
)
result = db.execute(stmt).scalars().all()
return result
第二处，数据库索引。原来的索引是单列idx_orders_status，但查询条件是WHERE status = ? ORDER BY created_at DESC，索引没覆盖排序字段，PostgreSQL要走filesort：
-- 优化前：单列索引
CREATE INDEX idx_orders_status ON orders(status);
-- 优化后：联合索引，覆盖排序
DROP INDEX idx_orders_status;
CREATE INDEX idx_orders_status_created ON orders(status, created_at DESC);
-- 执行计划对比
EXPLAIN ANALYZE SELECT * FROM orders WHERE status = 'PAID' ORDER BY created_at DESC LIMIT 50;
-- 优化前: Sort (cost=1234.56..1345.67 rows=4444 width=233) actual time=45.2ms
-- 优化后: Index Scan using idx_orders_status_created (cost=0.42..8.65 rows=50 width=233) actual time=0.08ms
执行计划从Sort变成了Index Scan，排序耗时从45ms降到0.08ms，这不是优化，这是降维打击。
4. 第三板斧：序列化优化——别让Pydantic背锅
4.1 问题定位
cProfile显示_serialize_orders占了241ms。看代码，问题在于：
# 优化前：用Pydantic v2的model_dump，但做了大量嵌套转换
class OrderOut(BaseModel):
id: int
status: str
created_at: datetime
items: list[ItemOut]
class Config:
from_attributes = True
# 每次请求创建50个OrderOut实例
return [OrderOut.model_validate(o) for o in orders]
Pydantic v2虽然快，但每次请求重复创建模型实例和datetime序列化都很重。而且model_validate内部会做字段校验、别名处理，这50个对象每个都要走一遍完整流程。
4.2 优化：手写序列化 + 缓存schema
优化思路有两个：
- 手写
to_dict()方法，跳过Pydantic的校验逻辑——反正数据来自数据库，类型是可信的。 - 缓存schema结构，避免重复构建。
# 优化后：手写轻量序列化
def serialize_order(order: Order) -> dict:
return {
"id": order.id,
"status": order.status,
"created_at": order.created_at.isoformat(), # 手动转ISO格式
"items": [
{"name": item.name, "price": item.price, "qty": item.qty}
for item in order.items
],
}
# 路由里直接调用
return [serialize_order(o) for o in orders]
这里有个小坑：datetime.isoformat()比Pydantic的序列化快3倍，因为Pydantic内部还要处理时区转换。而我们数据库存的TIMESTAMP WITHOUT TIME ZONE，直接取字符串就完事了。
5. 第四板斧：缓存策略——Redis与lru_cache双管齐下
数据库优化后，P95降到120ms左右。但还不够，因为同样的查询参数（比如看第一页的PAID订单）会被反复请求。这时候上缓存。
5.1 Redis缓存：面向重复查询
# cache.py
import redis.asyncio as redis
import json
redis_client = redis.from_url("redis://localhost:6379/0", decode_responses=True)
CACHE_TTL = 300 # 5分钟
async def get_cached_orders(status: str, page: int):
cache_key = f"orders:{status}:{page}"
cached = await redis_client.get(cache_key)
if cached:
return json.loads(cached)
return None
async def set_cached_orders(status: str, page: int, data: list):
cache_key = f"orders:{status}:{page}"
await redis_client.setex(cache_key, CACHE_TTL, json.dumps(data))
注意：缓存key必须包含所有查询参数。这里踩过一个坑，一开始只用了status做key，结果不同页返回了相同数据，线上事故。
5.2 lru_cache：热点参数内存缓存
对于更热的参数（比如status=PAID&page=1），Redis网络IO也是开销。用functools.lru_cache做进程内缓存：
from functools import lru_cache
@lru_cache(maxsize=64)
def get_orders_sync(status: str, page: int, page_size: int = 50):
# 注意：lru_cache的参数必须是可哈希的，page_size用默认值
stmt = (
select(Order)
.where(Order.status == status)
.options(selectinload(Order.items))
.order_by(Order.created_at.desc())
.offset((page - 1) * page_size)
.limit(page_size)
)
...
# 在路由中，优先查内存缓存
@app.get("/api/v1/orders")
async def get_orders_api(status: str, page: int = 1, page_size: int = 50):
# 先查内存缓存
cache_key = (status, page, page_size)
if cache_key in get_orders_sync.cache: # 直接访问cache属性
return get_orders_sync(*cache_key)
# 再查Redis
redis_data = await get_cached_orders(status, page)
if redis_data:
return redis_data
# 最后查数据库
data = get_orders_sync(status, page, page_size)
await set_cached_orders(status, page, data)
return data
版本坑提醒：lru_cache在Python 3.11里没问题，但如果你用functools.cache（3.9+），注意它的cache_clear()方法在并发下会抛异常。这里我选lru_cache是因为它线程安全（内部有锁）。
6. 压测数据对比——用数字说话
使用wrk压测，参数固定：wrk -t4 -c100 -d30s http://localhost:8000/api/v1/orders?status=PAID&page=2，每次压测前清空Redis缓存。
6.1 各阶段数据
| 阶段 | P50 (ms) | P95 (ms) | P99 (ms) | 吞吐量 (req/s) |
|---|---|---|---|---|
| 初始版本 | 198 | 312 | 458 | 320 |
| +selectinload & 联合索引 | 67 | 124 | 180 | 840 |
| +手写序列化 | 41 | 72 | 110 | 1150 |
| +Redis & lru_cache | 22 | 38 | 55 | 1480 |
关键解读：
- 数据库优化贡献了60%的延迟下降，这是最大的杠杆。
- 序列化优化贡献了25%，别小看Python对象的创建开销。
- 缓存贡献了剩下的15%，但对于热点数据，P99从110ms降到55ms，质变。
6.2 资源占用对比
数据库CPU使用率从70%降到25%（主要是消除了N+1和filesort）。Redis内存占用约12MB（缓存了2000个key），完全可接受。
7. 踩坑与总结
7.1 三个坑
-
selectinload不是银弹：如果明细表数据量极大（单订单上千item），selectinload生成的WHERE id IN (...)可能会超过PostgreSQL的参数上限（65535）。解决：分页加载或改用subqueryload。 -
缓存穿透：如果查询参数是随机的（比如带时间戳），缓存命中率会很低。我们加了布隆过滤器来拦截不存在的key，但这次业务场景没用到。
-
lru_cache的内存泄漏：maxsize=64看似安全，但如果status字段是用户可控的，攻击者可以构造大量不同status来撑爆内存。生产环境一定要限制status的枚举值。
7.2 总结
调优的本质是用数据代替直觉。这次经历告诉我：
- 先profile，再优化。cProfile和py-spy是必备工具。
- SQLAlchemy的懒加载是性能杀手，默认开启
selectinload。 - 联合索引要覆盖
WHERE和ORDER BY两个字段。 - Pydantic方便但重，热点接口手写序列化。
- 缓存是最后的手段，先优化代码本身。
最后留个思考题：如果这个接口的查询条件变成了范围查询（比如created_at > ?），缓存策略该怎么调整？欢迎评论区讨论。
附录：复现命令
# 安装依赖
pip install fastapi==0.104.1 sqlalchemy==2.0.23 uvicorn==0.24.0 redis==5.0.1
# 压测
wrk -t4 -c100 -d30s 'http://localhost:8000/api/v1/orders?status=PAID&page=2'
# 火焰图
py-spy record -o profile.svg --pid --duration 30
