一、问题背景：一个"看起来很简单"的接口
事情是这样的。我们有个订单查询接口 /api/v1/orders，逻辑说白了就是：根据用户ID查订单列表，每条订单再带上商品信息和用户昵称。
代码写得很"优雅"，用了SQLAlchemy的relationship，读起来很舒服：
@app.get("/api/v1/orders")
async def list_orders(user_id: int, db: Session = Depends(get_db)):
orders = db.query(Order).filter(Order.user_id == user_id).all()
return [OrderSchema.from_orm(o) for o in orders]
本地测试数据量小的时候，响应时间20ms都不到，测都没测就上线了。
然后线上就炸了。监控显示P99到了1.2s，QPS刚过50，数据库CPU直接拉满，接口开始大面积超时。用户投诉、老板追问，典型的"周五下午上线，周六凌晨回滚"剧本。
这篇文章就是记录我是怎么把它从1.2s干到85ms的。
二、环境与版本
先把环境交代清楚，不然数据没法复现：
- Python 3.11.6
- FastAPI 0.109.0
- Uvicorn 0.27.0（
--workers 4） - SQLAlchemy 2.0.25（同步ORM）
- PostgreSQL 15.4
- Redis 7.2.3
- py-spy 0.3.14
- locust 2.20.0（压测）
- 机器：4C8G，接口和DB分开部署
三、定位瓶颈：py-spy + 慢查询日志
3.1 先上py-spy看火焰图
py-spy的好处是不用改代码、不用重启，直接attach到进程：
pip install py-spy
py-spy record -o profile.svg --pid $(pgrep -f "uvicorn") --duration 30
压测跑起来，采样30秒，打开火焰图一看——80%的时间卡在 psycopg2 的 execute 上。这说明瓶颈不在Python代码，而在数据库往返。
3.2 打开SQLAlchemy慢查询日志
光知道卡在DB还不够，得看具体是哪条SQL。SQLAlchemy开echo太吵，用事件钩子更精准：
import time
from sqlalchemy import event
from sqlalchemy.engine import Engine
@event.listens_for(Engine, "before_cursor_execute")
def before_cursor_execute(conn, cursor, statement, parameters, context, executemany):
conn.info.setdefault("query_start_time", []).append(time.perf_counter())
@event.listens_for(Engine, "after_cursor_execute")
def after_cursor_execute(conn, cursor, statement, parameters, context, executemany):
total = time.perf_counter() - conn.info["query_start_time"].pop()
if total > 0.05: # 超过50ms记录
logger.warning(f"SLOW SQL ({total*1000:.1f}ms): {statement[:200]}")
跑了一次请求，日志直接刷屏：一个用户有50条订单，接口发了 1（主查询）+ 50（商品）+ 50（用户）= 101条SQL。经典的N+1问题。
而且主查询 SELECT * FROM orders WHERE user_id = ? 也慢，EXPLAIN ANALYZE 一看，orders表的user_id字段压根没索引，全表扫描。
四、方案设计与核心实现
定位清楚后，优化分三步走：
- 消N+1：用
selectinload预加载关联对象 - 加索引：给高频查询字段建复合索引
- 上缓存：热点用户订单列表进Redis，短TTL
4.1 消除N+1：selectinload替代懒加载
SQLAlchemy 2.0推荐用 selectinload，它会把关联查询合并成 IN (...) 一条SQL，比 joinedload 更适合一对多：
from sqlalchemy.orm import selectinload
@app.get("/api/v1/orders")
async def list_orders(user_id: int, db: Session = Depends(get_db)):
stmt = (
db.query(Order)
.options(
selectinload(Order.product),
selectinload(Order.user),
)
.filter(Order.user_id == user_id)
.order_by(Order.created_at.desc())
.limit(100)
)
orders = stmt.all()
return [OrderSchema.from_orm(o) for o in orders]
SQL从101条降到3条：主查询+商品IN查询+用户IN查询。
4.2 索引优化
看下最频繁的查询模式：WHERE user_id = ? ORDER BY created_at DESC LIMIT 100。
单列索引user_id不够，因为还要排序。建复合索引：
CREATE INDEX CONCURRENTLY idx_orders_user_created
ON orders (user_id, created_at DESC);
CONCURRENTLY 很重要，线上建索引不锁表。建完 EXPLAIN ANALYZE 从 Seq Scan 变成 Index Scan，主查询从320ms降到4ms。
4.3 Redis缓存热点数据
订单列表这种读多写少的数据，非常适合缓存。策略：
- Key：
orders:user:{user_id}:limit:{limit} - Value：JSON序列化后的响应体
- TTL：60秒（订单可能随时新增）
- 缓存穿透：空结果也缓存，TTL设10秒
- 缓存击穿：加互斥锁或随机TTL抖动
核心实现：
import json
import random
import redis.asyncio as aioredis
from fastapi import FastAPI, Depends, Query
app = FastAPI()
redis_client = aioredis.from_url("redis://localhost:6379/0", decode_responses=True)
CACHE_TTL = 60
async def get_orders_cached(user_id: int, limit: int = 100):
cache_key = f"orders:user:{user_id}:limit:{limit}"
cached = await redis_client.get(cache_key)
if cached is not None:
return json.loads(cached)
# 回源查询（同步ORM，包在线程池里）
orders = await fetch_orders_from_db(user_id, limit)
payload = [OrderSchema.from_orm(o).model_dump(mode="json") for o in orders]
# TTL加抖动，避免同一时刻大批key同时过期
ttl = CACHE_TTL + random.randint(-10, 10)
await redis_client.setex(cache_key, ttl, json.dumps(payload))
return payload
@app.get("/api/v1/orders")
async def list_orders(user_id: int, limit: int = Query(100, le=200)):
return await get_orders_cached(user_id, limit)
写操作记得清缓存，不然用户下了单看不到：
async def invalidate_user_orders(user_id: int):
# 用scan_iter避免keys命令阻塞
async for key in redis_client.scan_iter(match=f"orders:user:{user_id}:*"):
await redis_client.delete(key)
五、踩坑与优化
坑1：selectinload 在数据量大时反而慢。 一个用户有5000条订单时，IN (5000个id) 的SQL会超长。解决：业务层限制单次最多返回200条，配合分页。
坑2：Redis连接池没配好，高并发下报 ConnectionError。 默认连接池太小。改成：
redis_client = aioredis.from_url(
"redis://localhost:6379/0",
decode_responses=True,
max_connections=200,
socket_timeout=0.5,
socket_connect_timeout=0.5,
)
坑3：SQLAlchemy连接池耗尽。 默认 pool_size=5, max_overflow=10，QPS一上来就 TimeoutError。调到：
engine = create_engine(
DATABASE_URL,
pool_size=20,
max_overflow=30,
pool_pre_ping=True,
pool_recycle=1800,
)
pool_pre_ping=True 避免拿到已被DB断开的死连接。
坑4：Uvicorn workers数量。 一开始设了8，反而更慢——CPU 4核，超过核数的worker只会加剧上下文切换。改成 --workers 4，反而稳。
六、效果数据
Locust压测，100并发，持续5分钟，对比优化前后：
| 指标 | 优化前 | 优化后 | 提升 |
|---|---|---|---|
| P50 | 480ms | 32ms | 15x |
| P95 | 980ms | 68ms | 14.4x |
| P99 | 1200ms | 85ms | 14.1x |
| QPS | 52 | 680 | 13x |
| DB CPU | 98% | 22% | - |
| SQL数/请求 | 101 | 3（未命中缓存）/ 0（命中） | - |
缓存命中率稳定在 87% 左右（订单是读多写少场景）。DB负载直接降了一个数量级，现在这套架构撑个1000 QPS问题不大。
七、总结
这次调优其实没什么黑科技，就是老老实实按流程走：
- 先测量再优化，py-spy火焰图是神器，别凭感觉猜
- N+1是API性能头号杀手，
selectinload一定要用 - 索引要和查询模式匹配，
WHERE + ORDER BY就上复合索引 - 缓存不是银弹，TTL抖动、缓存穿透、写后失效这些细节必须处理
- 连接池参数不能全用默认值，DB和Redis都要按并发量调
最后一句忠告：上线前一定要压测。本地20ms和线上1.2s之间，差的不是代码，是数据量和并发。别学我周五上线。
代码都在上面了，有需要的同学直接抄。如果有更好的优化思路，欢迎评论区交流。
