一、问题背景
去年底接手一个订单服务，对外提供 /api/orders 查询接口。上线初期量不大，没在意性能。直到某次大促前压测，发现单个接口在 200 并发下 P99 直接飙到 800ms，QPS 卡在 120 左右上不去，数据库（PostgreSQL 15）的 CPU 使用率长期在 70%~85% 之间抖动。
业务方要求：单接口 P99 --duration 30 --rate 100
火焰图一眼就能看出问题：`serialize_order` 下面挂着一长串 `get_customer`、`get_items`，每个订单都单独查一次客户和明细——典型的 **N+1 查询**。
用 cProfile 在代码里埋点确认：
```python
import cProfile
import pstats
from io import StringIO
def profile_endpoint():
pr = cProfile.Profile()
pr.enable()
# 模拟单次请求
asyncio.run(fetch_orders(user_id=123))
pr.disable()
s = StringIO()
ps = pstats.Stats(pr, stream=s).sort_stats("cumulative")
ps.print_stats(20)
print(s.getvalue())
输出里 sqlalchemy/orm/loading.py 相关调用占了 60% 以上，实锤。
4.2 数据库查询优化：干掉 N+1
原始代码（简化）：
# 问题代码：每个订单都触发额外查询
@app.get("/api/orders")
async def list_orders(user_id: int, limit: int = 20):
async with async_session() as session:
result = await session.execute(
select(Order).where(Order.user_id == user_id).limit(limit)
)
orders = result.scalars().all()
# 下面这两行在循环里，每次访问触发 lazy load
return [
{
"id": o.id,
"customer": o.customer.name,
"items": [i.name for i in o.items],
}
for o in orders
]
改成 selectinload 预加载：
from sqlalchemy.orm import selectinload
@app.get("/api/orders")
async def list_orders(user_id: int, limit: int = 20):
async with async_session() as session:
stmt = (
select(Order)
.where(Order.user_id == user_id)
.options(
selectinload(Order.customer),
selectinload(Order.items),
)
.order_by(Order.created_at.desc())
.limit(limit)
)
result = await session.execute(stmt)
orders = result.scalars().all()
return [
{
"id": o.id,
"customer": o.customer.name,
"items": [i.name for i in o.items],
}
for o in orders
]
这一改，单请求的 SQL 数量从 1 + N + N 降到 3（主查询 + 两个 IN 查询），P99 从 800ms 掉到 320ms 左右。
4.3 缓存策略：Redis + 本地缓存
数据库优化后还有 320ms，离目标 150ms 差得远。分析发现：订单列表变化不频繁，但读非常频繁，典型读多写少，适合缓存。
策略：
- Redis 缓存：key 为
orders:{user_id}:{limit}，TTL 60s，序列化用orjson。 - 本地缓存：用
cachetools.TTLCache做一级缓存，TTL 5s，减少 Redis 网络往返。 - 缓存击穿：用
asyncio.Lock做单飞（singleflight），避免同一 key 并发穿透。
代码：
import orjson
from cachetools import TTLCache
from redis.asyncio import Redis
redis = Redis.from_url("redis://localhost:6379/0", decode_responses=False)
local_cache = TTLCache(maxsize=10000, ttl=5)
_locks: dict[str, asyncio.Lock] = {}
async def get_orders_cached(user_id: int, limit: int):
key = f"orders:{user_id}:{limit}"
# 一级：本地缓存
if key in local_cache:
return local_cache[key]
# 二级：Redis
cached = await redis.get(key)
if cached:
data = orjson.loads(cached)
local_cache[key] = data
return data
# 单飞，防击穿
lock = _locks.setdefault(key, asyncio.Lock())
async with lock:
# 双检
cached = await redis.get(key)
if cached:
return orjson.loads(cached)
data = await fetch_orders_from_db(user_id, limit)
payload = orjson.dumps(data)
await redis.set(key, payload, ex=60)
local_cache[key] = data
return data
Flask 侧要注意：Flask 是同步的，redis-py 用同步客户端，本地缓存用 cachetools.TTLCache 即可，但 gunicorn 多 worker 下本地缓存不共享，TTL 要设短一点（我设 3s），否则数据一致性问题会很难查。
4.4 连接池调优
FastAPI + asyncpg 默认池大小是 5，压测时明显不够。调整为：
engine = create_async_engine(
DATABASE_URL,
pool_size=20,
max_overflow=10,
pool_pre_ping=True,
pool_recycle=1800,
echo=False,
)
配合 pgbouncer 的 default_pool_size=25、max_client_conn=200。这里踩过一个坑：pgbouncer 用 transaction 模式时，asyncpg 的 prepared statement 会冲突，必须加 statement_cache_size=0：
engine = create_async_engine(
DATABASE_URL,
connect_args={"statement_cache_size": 0},
...
)
不加的话会报 prepared statement "__asyncpg_stmt_1__" already exists，压测时随机出现，非常难排查。
五、踩坑与优化
- py-spy 对 async 采样不准：早期用 py-spy 看 FastAPI，协程切换导致火焰图有点乱。后来配合 cProfile +
asyncio的 debug 模式一起看才准。 - selectinload 不是万能：如果 items 数量巨大（单订单上千条），
selectinload的 IN 查询会很大。这种情况改用joinedload或者分页加载明细。 - Redis 大 key：一开始 key 没设 limit 维度，
orders:{user_id}缓存了全部订单，单 key 几 MB。后来加上 limit 维度，并限制单次最多返回 100 条。 - 本地缓存不一致：Flask 多 worker 下本地缓存 TTL 太长会导致读到旧数据，改成 3s，并在写接口主动
local_cache.clear()。 - 压测工具本身成瓶颈：wrk 单机跑不满，后来用 3 台压测机 + Locust 分布式才压出真实数据。
六、效果数据
压测条件：200 并发，持续 5 分钟，请求 /api/orders?user_id=xxx&limit=20。
| 阶段 | P50 | P95 | P99 | QPS | DB CPU |
|---|---|---|---|---|---|
| 优化前 | 210ms | 620ms | 800ms | 120 | 78% |
| selectinload 后 | 95ms | 240ms | 320ms | 310 | 45% |
| + Redis 缓存 | 28ms | 65ms | 110ms | 560 | 22% |
| + 本地缓存 + 连接池 | 18ms | 42ms | 90ms | 680 | 15% |
最终 P99 90ms，QPS 680，超额完成目标。缓存命中率稳定在 94% 左右（Redis INFO 的 keyspace_hits / (keyspace_hits + keyspace_misses)）。
Flask 侧同步版本因为没法用 async 连接池，QPS 略低，最终在 520 左右，P99 约 130ms，但也够用了。
七、总结
这次调优的核心其实就三件事：
- 先 profiling 再动手，别凭感觉优化。py-spy + cProfile + pg_stat_statements 三件套足够定位 90% 的问题。
- N+1 是 API 性能头号杀手，SQLAlchemy 的
selectinload/joinedload一定要用对。 - 缓存要分层，Redis + 本地缓存组合能显著降低 P99，但要注意一致性、击穿和大 key。
最后提醒一句：连接池参数一定要和 pgbouncer 对齐，asyncpg 的 prepared statement 坑我踩了整整一天。希望这篇能帮你少走点弯路。
