一、问题背景：上线第二天就被运维拉群
事情是这样的。我们有个订单查询API，FastAPI + SQLAlchemy 2.0 + asyncpg + PostgreSQL 15，部署在K8s里3个Pod。上线第二天，运维甩了张监控图到群里：P99延迟3.7秒，数据库CPU 92%，QPS才180就扛不住了。
我第一反应是“代码没问题，是别人写的有问题”。结果查完代码，脸被打肿——是我自己上周写的。
这个API的逻辑很简单：根据用户ID查最近的20条订单，返回订单号和商品列表。但问题就藏在“商品列表”里。
先看下环境版本，避免你们说“版本不同没法复现”：
Python 3.11.4
FastAPI 0.104.1
SQLAlchemy 2.0.23
asyncpg 0.29.0
PostgreSQL 15.3 (Docker)
Redis 7.2 (Docker)
二、第一步：用profiling工具定位瓶颈，别瞎猜
很多人一上来就加缓存、加索引，但没数据支撑的优化都是玄学。我用了两把刀：
刀1：cProfile（函数级CPU分析）
写了个脚本直接压测接口，用cProfile抓函数调用耗时：
import cProfile
import pstats
import asyncio
from httpx import AsyncClient
async def bench():
async with AsyncClient(base_url="http://localhost:8000") as client:
for _ in range(50): # 压50次
await client.get("/api/orders?user_id=123&limit=20")
profiler = cProfile.Profile()
profiler.enable()
asyncio.run(bench())
profiler.disable()
stats = pstats.Stats(profiler)
stats.sort_stats("cumulative").print_stats(30) # 打印累计耗时前30
输出关键那几行（我简化了）：
ncalls tottime cumtime filename:lineno
50 0.012 3.210 api.py:30 get_orders # 总耗时3.2s/50次 ≈ 64ms/次
50 0.008 2.845 db.py:110 fetch_orders # 数据库操作占88%
250 0.156 1.920 db.py:140 fetch_items # 每个订单查一次商品
50 0.003 0.210 serializers.py:45 serialize # 序列化占6.5%
结论1：250次fetch_items调用，每次耗时7.7ms——这就是N+1查询。
刀2：py-spy（采样分析，线上用）
本地能跑cProfile，但线上不能随便挂。用py-spy dump看线程栈，确认是不是卡在数据库I/O：
# 找到PID
kubectl exec -it pod-name -- py-spy dump --pid 1
看到栈顶是asyncpg.protocol.prepare和await cursor.fetch，确认不是GIL问题，是数据库查询慢。
三、数据库查询优化：干掉N+1，从64ms降到9ms
问题根因：SQLAlchemy 2.0的lazy='subquery'在异步模式下会发额外查询。看代码：
# 优化前：每个order的items懒加载
async def fetch_orders(user_id: int, limit: int):
async with async_session() as session:
result = await session.execute(
select(Order)
.where(Order.user_id == user_id)
.order_by(Order.created_at.desc())
.limit(limit)
)
orders = result.scalars().unique().all()
# 这里访问 order.items 时，每个触发一次SELECT
return [{"id": o.id, "items": [i.name for i in o.items]} for o in orders]
优化方案：用selectinload一次性预加载items，同时加index：
from sqlalchemy.orm import selectinload
# 优化后：用selectinload预加载
async def fetch_orders_fixed(user_id: int, limit: int):
async with async_session() as session:
result = await session.execute(
select(Order)
.options(selectinload(Order.items)) # 关键：一次IN查询
.where(Order.user_id == user_id)
.order_by(Order.created_at.desc())
.limit(limit)
)
orders = result.scalars().unique().all()
return [{"id": o.id, "items": [i.name for i in o.items]} for o in orders]
同时给orders.user_id和orders.created_at建了复合索引：
CREATE INDEX idx_orders_user_created ON orders (user_id, created_at DESC);
效果：接口单次调用从64ms降到9ms（数据库查询部分从56ms降到4ms）。但离我目标还差得远，因为QPS 200时还是扛不住——问题在重复查询。
四、缓存策略：Redis二级缓存 + 本地LRU
数据库优化后，P99仍有1.2s（因为QPS高时连接池打满）。接下来上缓存，我做了两层：
- L1本地缓存：
functools.lru_cache，TTL 5秒。适合热用户，避免每次请求都走Redis网络。 - L2 Redis缓存：TTL 60秒。所有Pod共享，解决多实例一致性。
实现代码如下：
import json
import asyncio
from functools import lru_cache
from redis.asyncio import Redis
redis = Redis.from_url("redis://redis:6379/0", decode_responses=True)
# 本地缓存装饰器（带TTL）
def ttl_lru_cache(ttl: int, maxsize: int = 128):
def decorator(fn):
cache = {}
lock = asyncio.Lock()
@lru_cache(maxsize=maxsize)
def _wrapped(*args):
return (asyncio.get_event_loop().time(), fn(*args))
async def wrapper(*args, **kwargs):
key = (args, tuple(sorted(kwargs.items())))
now = asyncio.get_event_loop().time()
async with lock:
if key in cache:
ts, val = cache[key]
if now - ts < ttl:
return val
val = await fn(*args, **kwargs)
cache[key] = (now, val)
return val
return wrapper
return decorator
# 实际使用：优先Redis，再本地
async def get_orders_cached(user_id: int, limit: int):
cache_key = f"orders:{user_id}:{limit}"
# 1. 查Redis
cached = await redis.get(cache_key)
if cached:
return json.loads(cached)
# 2. 查数据库（走上面优化过的fetch_orders_fixed）
data = await fetch_orders_fixed(user_id, limit)
# 3. 写Redis，TTL 60秒
await redis.setex(cache_key, 60, json.dumps(data))
return data
踩坑记录：
- 千万别直接
json.dumpsSQLAlchemy对象，会报Object of type Order is not JSON serializable。必须先把ORM转dict。 - Redis连接池默认100，压测时不够，调到500：
Redis(..., max_connections=500)。 - 本地缓存必须加
asyncio.Lock，否则并发下重复查库。
五、压测效果数据：P99从3.7s到180ms
用locust压测，模拟真实场景：用户ID均匀分布（1000个用户），每个用户查20条订单。4核8G Pod，3个副本：
| 指标 | 优化前 | 优化后 | 提升 |
|---|---|---|---|
| QPS（稳定） | 180 | 940 | 5.2倍 |
| P99延迟 | 3.7s | 180ms | 20.5倍 |
| 数据库CPU | 92% | 31% | -66% |
| Redis内存 | 0MB | 78MB | 新增 |
压测命令（locustfile.py关键部分）：
from locust import HttpUser, task, between
import random
class OrderUser(HttpUser):
wait_time = between(0.1, 0.3)
@task(2)
def get_orders(self):
user_id = random.randint(1, 1000)
self.client.get(f"/api/orders?user_id={user_id}&limit=20")
@task(1)
def get_orders_cached(self):
# 模拟热用户重复访问
user_id = random.randint(1, 50)
self.client.get(f"/api/orders?user_id={user_id}&limit=20")
最终结论：
- 数据库优化解决的是“慢查询”问题，收益巨大（64ms→9ms）。
- 缓存解决的是“重复计算”问题，收益更大（QPS提升5倍）。
- 如果只做不做profiling，你可能会去调数据库参数，白白浪费时间。
六、总结与反思
- 先profile再动手：cProfile、py-spy、EXPLAIN ANALYZE，这三个工具比任何“经验”都靠谱。
- N+1查询是FastAPI+SQLAlchemy最常见的坑：异步模式下尤其严重，
selectinload是首选。 - 缓存不是银弹：缓存击穿、雪崩、一致性。我们只做了缓存穿透防护（空值也缓存），没做击穿防护（加锁），因为QPS还不到1000。如果QPS上万，必须上
singleflight。 - 版本推荐：如果你用Flask，同样的思路适用，但异步性能和并发会差一截。FastAPI + asyncpg + selectinload是唯一能跑满PostgreSQL的连接池的方案。
最后说句实话：这个接口如果当初设计时就把items放进一个JSONB字段，根本不会有N+1问题。但现实业务里，订单和商品毕竟是两张表。所以，调优的本质是理解你的数据和访问模式，工具只是加速器。
如果你们也有类似问题，评论区聊聊你的P99是多少？我先来：现在我的P99稳定在180ms，QPS 940，数据库CPU 31%。
