一、问题背景
事情是这样的。我们有一个内部数据服务，主框架是 FastAPI（0.110.0），对外提供设备状态查询接口。接口本身逻辑不复杂：根据 device_id 查设备信息，再关联查最近的状态记录和告警列表。
上线初期数据量小，响应一直在 200ms 左右，没人关注。但业务接入方越来越多，单设备的关联记录涨到几千条之后，接口开始拉胯：监控上看 P95 到了 2.3s，Prometheus 里 http_request_duration_seconds 的 99 分位直接爆表。更麻烦的是压测一上 50 QPS，就出现大面积 502，Gunicorn 的 worker 全被慢请求占满。
这篇文章就是我把这个过程从头到尾记下来，包括走了哪些弯路。
二、环境与版本
先把环境交代清楚，不然数据没法复现：
- Python 3.11.6
- FastAPI 0.110.0 + Uvicorn 0.27.1
- SQLAlchemy 2.0.28（async 模式，asyncpg 0.29.0）
- PostgreSQL 15.4
- Redis 7.2.4（redis-py 5.0.1）
- Gunicorn 21.2.0，4 workers，
uvicorn.workers.UvicornWorker - 压测工具：wrk 4.2.0 + locust 2.24.0
- 机器：4C8G 容器，数据库独立实例
部署方式是 gunicorn -k uvicorn.workers.UvicornWorker -w 4 -b 0.0.0.0:8000 app:app，没上 Kubernetes，就一台机器。
三、方案设计：先定位，再动手
我给自己定了个原则：没 profiling 数据之前不改一行代码。凭感觉优化是最容易白干活的。
排查路径分三步：
- CPU/调用栈层面：用
py-spy采样，看时间花在哪 - SQL 层面：打开 SQLAlchemy 的 echo，统计单次请求的 SQL 条数
- 缓存层面：判断哪些数据能缓存、缓存粒度多大、失效策略怎么定
工具选型上，py-spy 是无侵入的，生产环境也能跑；cProfile 精度更高但开销大，只在本地复现时用。
四、核心实现
4.1 用 py-spy 抓现场
线上直接采样 30 秒：
py-spy record -o profile.svg --pid $(pgrep -f "uvicorn") --duration 30 --rate 100
生成的火焰图一看就明白了：超过 60% 的时间卡在 asyncpg 的 fetch 上，还有约 15% 在 Pydantic 的 model_validate。基本可以断定是数据库查询问题。
4.2 复现并数 SQL 条数
本地写了个最小复现脚本，打开 SQL echo：
# reproduce.py
import asyncio
import time
from sqlalchemy import select, func
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from models import Device, StatusRecord, Alert
engine = create_async_engine(
"postgresql+asyncpg://user:pwd@localhost:5432/devdb",
echo=True, # 打开 SQL 日志
pool_size=20,
max_overflow=10,
)
Session = async_sessionmaker(engine, expire_on_commit=False)
async def get_device_detail(device_id: int):
async with Session() as session:
# 第一层：查设备
device = await session.get(Device, device_id)
# 第二层：查状态记录（这里就是 N+1 的坑）
records = (await session.execute(
select(StatusRecord).where(StatusRecord.device_id == device_id)
)).scalars().all()
# 第三层：每条状态记录再查告警
for r in records:
alerts = (await session.execute(
select(Alert).where(Alert.record_id == r.id)
)).scalars().all()
r.alerts = alerts
return device, records
async def main():
t0 = time.perf_counter()
device, records = await get_device_detail(1024)
print(f"records={len(records)}, elapsed={time.perf_counter()-t0:.3f}s")
if __name__ == "__main__":
asyncio.run(main())
跑一次输出：records=2847, elapsed=2.31s，SQL 日志刷了 2849 条。典型的 N+1，每条状态记录单独查一次告警。
4.3 用 selectinload 改写查询
SQLAlchemy 2.0 的关系加载用 selectinload 最合适——它会把关联查询合并成 WHERE record_id IN (...) 的形式，一次拿完。
from sqlalchemy.orm import selectinload
from sqlalchemy import select
async def get_device_detail_optimized(device_id: int, limit: int = 50):
async with Session() as session:
stmt = (
select(Device)
.options(
selectinload(Device.records).selectinload(StatusRecord.alerts)
)
.where(Device.id == device_id)
)
device = (await session.execute(stmt)).scalar_one_or_none()
if device is None:
return None, []
# 只返回最近 limit 条，避免全量序列化
records = sorted(device.records, key=lambda r: r.created_at, reverse=True)[:limit]
return device, records
注意这里还有个细节：原逻辑返回全量 2847 条，光 Pydantic 序列化就要 400ms+。业务上其实只展示最近 50 条，所以顺手加了 limit。这一步比 SQL 优化省得还多。
4.4 加两级缓存
设备基础信息变化极少，适合缓存。我用了进程内 LRU + Redis 两层：
import json
import hashlib
from functools import lru_cache
from redis.asyncio import Redis
redis = Redis(host="localhost", port=6379, decode_responses=True)
def _key(device_id: int) -> str:
return f"device:detail:v1:{device_id}"
async def get_device_detail_cached(device_id: int):
key = _key(device_id)
cached = await redis.get(key)
if cached:
return json.loads(cached)
device, records = await get_device_detail_optimized(device_id)
if device is None:
return None
payload = {
"device": {"id": device.id, "name": device.name, "model": device.model},
"records": [
{
"id": r.id,
"status": r.status,
"created_at": r.created_at.isoformat(),
"alerts": [{"id": a.id, "level": a.level} for a in r.alerts],
}
for r in records
],
}
# TTL 60s，加随机抖动防止雪崩
import random
ttl = 60 + random.randint(0, 10)
await redis.setex(key, ttl, json.dumps(payload))
return payload
TTL 设 60 秒是因为状态记录每分钟会更新一次，再短没必要，再长业务方会说数据不新鲜。随机抖动是为了避免同一批 key 同时过期。
4.5 调整 Gunicorn 配置
4C 机器上 -w 4 其实是偏少的。FastAPI 是 IO 密集，worker 数可以开到 2 * CPU + 1 = 9。但别一次拉满，得配合压测看。最终用了：
gunicorn app:app \
-k uvicorn.workers.UvicornWorker \
-w 8 \
--max-requests 5000 \
--max-requests-jitter 500 \
--timeout 30 \
--graceful-timeout 30 \
-b 0.0.0.0:8000
--max-requests 是防止长时间运行后内存碎片，每 5000 个请求重启一个 worker。
五、踩坑与优化
坑一：selectinload 不是万能的。 我一开始对 Device.records 用了 joinedload，结果 JOIN 出 2800 行，反序列化时 SQLAlchemy 的 identity map 去重开销巨大，反而比 N+1 慢。一对多关系一定要用 selectinload。
坑二：缓存 key 没带版本号，灰度时踩过雷。 后来统一加 v1 前缀，字段结构变了就升版本，老 key 自然过期，不用手动清。
坑三：Pydantic v2 的 model_validate 对大批量数据有开销。 原来返回全量时，序列化占了近 20% 时间。现在我直接构造 dict 返回，绕开 Pydantic 的模型校验，只在入口做校验。这个 trade-off 需要你评估业务是否接受。
坑四：Gunicorn worker 从 4 加到 8 后，数据库连接数不够用了。 asyncpg 每个 worker 默认池 20，8 个 worker 就是 160 连接，超过了 Postgres 的 max_connections=100。最后把 pool_size 降到 10，max_overflow 到 5，总连接控制在 120 以内，同时把 Postgres 的 max_connections 提到 200。
六、效果数据
用 wrk 压测，命令：wrk -t4 -c100 -d60s http://localhost:8000/api/devices/1024
| 阶段 | P50 | P95 | P99 | QPS | 错误率 |
|---|---|---|---|---|---|
| 优化前 | 890ms | 2300ms | 3800ms | 52 | 3.1% |
| SQL 优化后 | 210ms | 480ms | 720ms | 190 | 0% |
| 加缓存后 | 45ms | 180ms | 260ms | 420 | 0% |
CPU 占用从优化前的 95%（持续打满）降到 40% 左右，内存稳定在 1.2G。数据库这边慢查询日志清空，pg_stat_statements 显示最耗时查询从 2100ms 降到 12ms。
缓存命中率稳定在 87%（60 秒 TTL 下的合理值）。
七、总结
回头看，这次调优的核心就三件事：
- 先测再改。
py-spy5 分钟就能告诉你瓶颈在哪，别瞎猜。 - N+1 是接口性能的头号杀手。SQLAlchemy 里优先用
selectinload，别用joinedload处理一对多。 - 缓存粒度要跟业务对齐。TTL 不是越短越好，要看数据更新频率和业务容忍度。
另外提一句，如果你们也在用 Flask 而不是 FastAPI，思路完全一样。Flask 用 flask-sqlalchemy 的 selectinload 写法几乎一致，profiling 也可以用 flask-profiler 或直接上 py-spy。唯一区别是 Flask 没有 async，worker 数可以开得更激进，但要注意数据库连接池别撑爆。
性能优化没有银弹，只有一个个具体的、可度量的改动。希望这篇记录能帮你少走点弯路。
