一、问题背景：一个"看起来很简单"的接口
上个月接手一个订单查询服务，接口逻辑简单到不能再简单：根据用户ID返回最近订单列表，附带商品名和状态。代码写得很"干净"，但上线后监控面板一直飘红：
- P50：180ms
- P95：620ms
- P99：820ms
- 压测QPS到120时，错误率开始抬头
产品经理的原话是"用户点一下要等一秒，这还能用？"。
我先做了两件事：一是把接口从Flask迁移到FastAPI（团队新服务统一用FastAPI），二是老老实实做profiling。事实证明，不测就优化等于瞎猜，我一开始以为是FastAPI的锅，结果跟框架半毛钱关系没有。
二、环境与版本
先把环境交代清楚，避免"你版本跟我一样吗"的扯皮：
| 组件 | 版本 |
|---|---|
| Python | 3.11.6 |
| FastAPI | 0.110.0 |
| Flask（旧服务） | 3.0.2 |
| SQLAlchemy | 2.0.28 |
| Redis | 7.2.4 |
| PostgreSQL | 15.6 |
| uvicorn | 0.27.1 |
| 压测工具 | wrk 4.2.0 + locust 2.24 |
机器配置：4C8G，数据库同机房，网络RTT --duration 30
火焰图确认了另一个问题：`requests.get()` 同步调用第三方状态服务，单次耗时80-120ms，**串行阻塞**在请求链路里。
### 4.2 数据库优化：批量查询 + 复合索引
原代码（问题版）：
```python
# 反面教材
@app.get("/api/orders")
def get_orders(user_id: int):
orders = db.query(Order).filter(Order.user_id == user_id).all()
result = []
for o in orders:
product = db.query(Product).filter(Product.id == o.product_id).first() # N+1
result.append({"id": o.id, "product": product.name, "status": o.status})
return result
改造后：
# app/crud.py
from sqlalchemy import select
from sqlalchemy.orm import joinedload
def get_orders_with_product(db, user_id: int, limit: int = 20):
stmt = (
select(Order)
.options(joinedload(Order.product))
.where(Order.user_id == user_id)
.order_by(Order.created_at.desc())
.limit(limit)
)
return db.execute(stmt).unique().scalars().all()
配合复合索引：
CREATE INDEX CONCURRENTLY idx_orders_user_created
ON orders (user_id, created_at DESC)
INCLUDE (product_id, status);
INCLUDE 让查询走覆盖索引，不用回表。这一步让单次查询从32ms降到4ms。
4.3 缓存策略：Redis + 本地LRU二级缓存
订单数据"读多写少"，非常适合缓存。但直接上Redis也有坑——热点key和缓存击穿。
我的方案是两级缓存：
# app/cache.py
import json
import redis
from cachetools import TTLCache
from functools import wraps
local_cache = TTLCache(maxsize=2000, ttl=10) # 本地10s
rds = redis.Redis(host="127.0.0.1", port=6379, decode_responses=True)
def two_level_cache(prefix: str, ttl: int = 60):
def decorator(func):
@wraps(func)
def wrapper(user_id: int, *args, **kwargs):
key = f"{prefix}:{user_id}"
# L1 本地
if key in local_cache:
return local_cache[key]
# L2 Redis
cached = rds.get(key)
if cached:
data = json.loads(cached)
local_cache[key] = data
return data
# 回源
data = func(user_id, *args, **kwargs)
rds.setex(key, ttl, json.dumps(data, default=str))
local_cache[key] = data
return data
return wrapper
return decorator
写入时主动失效：
def create_order(db, user_id, product_id):
order = Order(user_id=user_id, product_id=product_id)
db.add(order)
db.commit()
rds.delete(f"orders:{user_id}") # 清Redis
local_cache.pop(f"orders:{user_id}", None) # 清本地
return order
注意：本地缓存TTL只有10s，是为了在Redis失效瞬间兜底，避免缓存击穿冲垮数据库。多实例部署时本地缓存会有短暂不一致，但订单查询场景能接受。
4.4 异步改造：干掉同步阻塞
第三方状态服务改成 httpx.AsyncClient，并发调用：
import httpx
async def fetch_status(order_ids: list[int]) -> dict:
async with httpx.AsyncClient(timeout=2.0) as client:
tasks = [client.get(f"https://status.internal/{oid}") for oid in order_ids]
responses = await asyncio.gather(*tasks, return_exceptions=True)
return {oid: r.json() for oid, r in zip(order_ids, responses) if not isinstance(r, Exception)}
接口改成 async def，数据库用 AsyncSession（SQLAlchemy 2.0 原生支持）：
from sqlalchemy.ext.asyncio import AsyncSession
@app.get("/api/orders")
async def get_orders(user_id: int, db: AsyncSession = Depends(get_db)):
orders = await get_orders_with_product(db, user_id)
status_map = await fetch_status([o.id for o in orders])
return [{"id": o.id, "product": o.product.name,
"status": status_map.get(o.id, {}).get("status")} for o in orders]
五、踩坑与优化
坑1：joinedload + limit 会生成子查询，SQLAlchemy 2.0 下如果关联是一对多，limit 会作用在外层，导致数据量爆炸。我这里是多对一，所以安全；一对多要用 selectinload。
坑2：Redis连接池没配，默认 max_connections 是 2^31，但实际压测时连接数飙升到800+，触发系统文件描述符限制。改成：
pool = redis.ConnectionPool(
host="127.0.0.1", port=6379, max_connections=50,
socket_timeout=0.5, socket_connect_timeout=0.5,
health_check_interval=30
)
坑3：uvicorn worker数。一开始用默认1个worker，CPU只用了一个核。改成 --workers 4（等于CPU核数），QPS直接翻倍。
坑4：缓存序列化用 json.dumps(default=str) 会把datetime转成字符串，反序列化后前端拿到的是字符串不是时间戳。后来改成显式 isoformat()，前端自己解析。
六、效果数据
压测命令（wrk）：
wrk -t8 -c200 -d60s --latency \
"http://localhost:8000/api/orders?user_id=12345"
| 指标 | 优化前 | 优化后 | 提升 |
|---|---|---|---|
| P50 | 180ms | 12ms | 15x |
| P95 | 620ms | 38ms | 16x |
| P99 | 820ms | 47ms | 17x |
| QPS | 118 | 1420 | 12x |
| 错误率 | 2.3% | 0% | - |
| SQL次数/请求 | 3 | 1 | 3x |
数据库CPU从压测时的85%降到12%，Redis命中率稳定在96%（本地+Redis合计）。第三方状态服务因为并发调用，整体耗时从平均110ms降到35ms。
有个反直觉的点：本地缓存TTL从10s调到30s后，QPS反而略降。原因是本地缓存命中率上升，Redis压力下降，但本地缓存命中时延（~0.1ms）和Redis时延（~0.8ms）差异在整体链路里占比太小，反而因为TTL长导致更多请求打到"过期但未更新"的边界。最后调回10s。
七、总结
这次调优的核心经验就三条：
- 先profiling再动手。我一开始想当然觉得是FastAPI慢，结果cProfile一跑，全在SQL和HTTP上。
- N+1是API性能头号杀手。150次SQL压成1次，这一项贡献了60%的优化收益。
- 缓存要分级。本地+Redis的组合能扛住热点key，但TTL和一致性策略要结合业务容忍度来定。
最后提醒一句：别过早优化。这个接口上线初期QPS不到10，根本不需要缓存。是业务量涨上来了才暴露出问题。优化要有数据支撑，不然就是给自己加维护负担。
