一、问题背景：一个“看起来很简单”的查询接口
事情起因很普通。我们内部有一个商品聚合查询接口 /api/v1/products/search，逻辑就是：按关键词搜索商品，返回商品基本信息 + 所属类目 + 最近3条价格记录。业务方反馈“列表页转圈要两秒”，我一开始没太当回事，觉得加个索引就完事了。
结果上监控一看，确实不乐观：
- P50：420ms
- P95：1.8s
- P99：2.6s
- 平均QPS：42，压到100并发时错误率开始上升
- 数据库CPU在高峰期打到70%
这个接口同时被FastAPI（新版网关层）和Flask（老后台管理端）调用，两边都走同一套SQLAlchemy模型和Service层。所以优化不能只改一边，得从共用的数据访问层下手。
我的目标很明确：P95压到200ms以内，QPS至少翻5倍，数据库压力明显下降。
二、环境与版本：先把基线固定住
压测和profiling最怕环境漂移，先把版本钉死：
- Python 3.11.7
- FastAPI 0.110.0 + Uvicorn 0.29.0（workers=4）
- Flask 3.0.2 + Gunicorn 21.2.0（workers=4, worker_class=uvicorn.workers.UvicornWorker 用于对比）
- SQLAlchemy 2.0.29 + asyncpg 0.29.0（FastAPI侧异步）/ psycopg2-binary 2.9.9（Flask侧同步）
- PostgreSQL 15.6
- Redis 7.2.4
- 压测工具：wrk 4.2.0 + locust 2.24.0
- Profiling：cProfile、py-spy 0.3.14、SQLAlchemy echo、pg_stat_statements
机器配置：4C8G 容器，数据库独立实例 8C16G。压测固定 100 并发、持续 60s，取稳定段数据。
三、方案设计：先测量，再动手
我给自己定了三步：
- 定位：用 py-spy 抓火焰图，cProfile 看函数级耗时，pg_stat_statements 看SQL。
- 拆分瓶颈：把总耗时拆成“框架开销 / 业务逻辑 / 数据库 / 序列化”四块。
- 逐项优化：数据库查询优化 → 缓存策略 → 序列化与框架层微调。
先说结论：这个接口1.8s里，数据库相关占了约1.1s，序列化占了约300ms，框架和业务逻辑其实没那么重。也就是说，大头在数据访问层，不在FastAPI本身。很多人一上来就换框架、加worker，方向就错了。
3.1 Profiling：py-spy + cProfile
py-spy 对运行中的进程几乎无侵入，适合先看整体：
# 找到 uvicorn worker pid
py-spy top --pid 12345 --duration 30
# 生成火焰图
py-spy record -o profile.svg --pid 12345 --duration 60 --rate 100
火焰图里最显眼的是 sqlalchemy/orm/loading.py 和 asyncpg 相关调用反复出现，说明查询次数过多。再用 cProfile 在本地复现单次请求：
import cProfile
import pstats
from app.main import app
from fastapi.testclient import TestClient
client = TestClient(app)
def run():
for _ in range(20):
client.get("/api/v1/products/search", params={"q": "phone", "page": 1, "size": 20})
profiler = cProfile.Profile()
profiler.enable()
run()
profiler.disable()
stats = pstats.Stats(profiler).sort_stats("cumulative")
stats.print_stats(30)
输出里 load_scalar_attributes、_load_for_state 调用次数高得离谱。再看 pg_stat_statements：
SELECT query, calls, total_exec_time, mean_exec_time
FROM pg_stat_statements
ORDER BY total_exec_time DESC
LIMIT 10;
发现同一个 SELECT ... FROM categories WHERE id = $1 被调用了 20 次/请求，价格记录查询也是每条商品一次。典型 N+1。
四、核心实现：三处改动，收益最大
4.1 数据库查询优化：selectinload + 索引
原来的代码大概是这样（简化）：
# 优化前：N+1
products = session.execute(
select(Product).where(Product.name.ilike(f"%{q}%")).limit(20)
).scalars().all()
result = []
for p in products:
category = session.get(Category, p.category_id) # 每次一条SQL
prices = session.execute(
select(Price).where(Price.product_id == p.id)
.order_by(Price.created_at.desc()).limit(3)
).scalars().all() # 每条商品一次
result.append({"product": p, "category": category, "prices": prices})
改成 SQLAlchemy 2.0 的 selectinload，把 category 和 prices 一次性加载：
from sqlalchemy.orm import selectinload
from sqlalchemy import select
stmt = (
select(Product)
.where(Product.name.ilike(f"%{q}%"))
.options(
selectinload(Product.category),
selectinload(Product.prices), # 关系上配合 order_by 和 limit 需用窗口或子查询
)
.limit(20)
)
products = session.execute(stmt).scalars().all()
注意：selectinload 对“每个商品只取最近3条价格”这种带 limit 的场景不能直接生效。我的做法是把价格查询单独抽出来，用 IN + 窗口函数 一次拿回：
SELECT * FROM (
SELECT p.*, ROW_NUMBER() OVER (
PARTITION BY product_id ORDER BY created_at DESC
) AS rn
FROM prices p
WHERE p.product_id = ANY(:ids)
) t WHERE t.rn str:
raw = f"search:{q.lower().strip()}:{page}:{size}"
return "api:" + hashlib.md5(raw.encode()).hexdigest()
def get_cached(key: str):
data = r.get(key)
return json.loads(data) if data else None
def set_cached(key: str, value, ttl: int = 60):
# 抖动 ±10s，避免同一时刻大量key同时过期
jitter = random.randint(-10, 10)
r.setex(key, ttl + jitter, json.dumps(value, ensure_ascii=False))
在路由里：
@app.get("/api/v1/products/search")
async def search(q: str, page: int = 1, size: int = 20):
key = make_cache_key(q, page, size)
cached = await run_in_threadpool(get_cached, key) # 或异步redis客户端
if cached:
return cached
data = await query_products(q, page, size)
await run_in_threadpool(set_cached, key, data, 60)
return data
这里有个坑：一开始我用同步 redis 客户端直接在 async 路由里调用，把事件循环堵住了，QPS 反而下降。后来改成 redis.asyncio 或者丢到线程池，问题解决。
另外，缓存 key 一定要包含分页参数和排序参数，否则会串数据。我们上线前就遇到过 page=2 返回了 page=1 的结果，原因就是 key 少拼了 page。
4.3 序列化优化：orjson 替换默认 JSON
FastAPI 默认用 json 序列化，大列表下开销不小。换成 orjson：
from fastapi.responses import ORJSONResponse
from fastapi import FastAPI
app = FastAPI(default_response_class=ORJSONResponse)
Flask 侧用 flask-orjson 或手动 orjson.dumps。这一项单独测下来，20条商品的响应序列化从约 300ms 降到约 90ms。别小看序列化，数据量大时它很吃CPU。
五、踩坑与优化：那些让我加班到十点的细节
坑1：selectinload + limit 不生效。 前面说了，关系上带 limit 会退化成全量加载再截断。必须用窗口函数或单独查询。
坑2：缓存穿透。 有人用不存在的关键词疯狂刷接口，每次都打到数据库。加了空结果缓存（TTL 10s）和布隆过滤器后缓解。
坑3：Gunicorn worker 数量。 Flask 侧原来配了 8 个 worker，但容器只有 4C，上下文切换严重。改成 workers=2*CPU+1=9 是理论值，实际压测 4~5 个最稳。FastAPI 侧 Uvicorn workers=4 比较合适。参数不是越多越好。
坑4：连接池。 SQLAlchemy 默认 pool_size=5，max_overflow=10。高并发下连接等待明显。调整：
engine = create_async_engine(
DATABASE_URL,
pool_size=20,
max_overflow=10,
pool_pre_ping=True,
pool_recycle=1800,
)
PostgreSQL 侧 max_connections=200，配合 pgBouncer 更稳。
坑5：压测工具本身。 wrk 单机压的时候，客户端先到瓶颈了。后来用 locust 分布式压，数据才可信。别拿压测客户端的极限当服务端极限。
六、效果数据：优化前后对比
同一套压测脚本、同一台机器、100并发、60s：
| 指标 | 优化前 | 优化后 | 变化 |
|---|---|---|---|
| P50 | 420ms | 45ms | ↓89% |
| P95 | 1800ms | 120ms | ↓93% |
| P99 | 2600ms | 210ms | ↓92% |
| QPS | 42 | 680 | ↑16倍 |
| 数据库QPS | ~1200 | ~200 | ↓83% |
| 错误率 | 1.2% | 0% | - |
| 数据库CPU峰值 | 70% | 18% | ↓74% |
分项耗时（单请求，20条商品）：
- 数据库查询：1100ms → 210ms
- 序列化：300ms → 90ms
- 业务逻辑+框架：约400ms → 约120ms（含缓存命中时更低）
缓存命中率稳定在 78% 左右，热门关键词基本都命中。Redis 内存占用约 180MB，可接受。
Flask 老后台侧因为共用 Service 层，QPS 从 35 提升到 410，效果同样明显。
七、总结
这次调优最大的感受是：不要猜，要测。我一开始也以为是FastAPI慢，结果火焰图直接打脸——瓶颈在N+1查询和序列化。顺序应该是：profiling定位 → 数据库查询优化 → 缓存 → 序列化/框架微调。
几个可复用的经验：
py-spy+pg_stat_statements是定位API性能问题的黄金组合，成本低、信息量大。- SQLAlchemy 2.0 的
selectinload能解决大部分N+1，但带 limit 的关系要特殊处理。 - 缓存 key 必须包含所有影响结果的参数，TTL 加随机抖动。
- 异步框架里千万别混用同步IO，否则事件循环被堵，性能不升反降。
- 连接池、worker 数量这些参数要压测调，别照搬公式。
优化不是一劳永逸的。上线两周后我又加了一次监控告警，当 P95 超过 300ms 就触发，防止数据量增长后性能悄悄退化。性能这件事，得持续盯着。