一、问题背景
事情是这样的，我们有个内部商品查询接口 GET /api/v1/products/search，给运营后台和几个内部服务用。功能很简单：按关键词、类目、价格区间查商品列表，带分页，返回商品基础信息加库存和价格。
上线初期数据量小，没人管。后来商品表到了 320 万行，这个接口就开始不对劲了：
- P95 响应时间 1.8s，P99 到 3.2s
- 单机 QPS 超过 200 就开始出现 5s 超时
- 数据库 CPU 常年在 80% 以上，高峰期直接打满
运营那边天天投诉"搜个商品要等两秒"，我作为接口的负责人，只能硬着头皮上。这篇文章就是这次调优的完整记录，包括怎么定位、怎么改、改完到底快了多少。
先说结论：瓶颈不在 FastAPI 本身，也不在 Python 的 GIL，而在数据库——具体是 N+1 查询加上缺失的复合索引，再叠加一个没有任何缓存的同步查询链路。
二、环境与版本
先交代下环境，不然数据没法复现：
| 组件 | 版本/配置 |
|---|---|
| Python | 3.11.6 |
| FastAPI | 0.109.2 |
| Uvicorn | 0.27.1，4 workers |
| SQLAlchemy | 2.0.25（async + asyncpg 0.29.0） |
| PostgreSQL | 15.4，4C8G，shared_buffers=2GB |
| Redis | 7.2.4，单节点 |
| py-spy | 0.3.14 |
| Locust | 2.20.1 |
机器是 4 核 8G 的云主机，接口部署在 Uvicorn 4 workers 下。压测机和被测机分离，避免互相干扰。
三、方案设计：先定位，再动手
我的原则是：没有 profiling 数据之前，一行优化代码都不写。 凭直觉优化，十有八九是在优化一个不存在的瓶颈。
整体分三步：
- 定位：用 py-spy 做火焰图采样 + 自建中间件记录慢请求的分段耗时，确认时间到底花在哪。
- 优化数据库：根据慢查询日志和
EXPLAIN ANALYZE，处理 N+1 查询、补索引、改分页方式。 - 加缓存：对热点查询做 Redis 缓存，设计好 key 和失效策略。
先上定位工具。
3.1 py-spy 火焰图
py-spy 的好处是不用改代码、不重启进程，直接 attach 到运行中的 worker 上采样：
# 找到 uvicorn worker 的 pid
ps -ef | grep uvicorn
# 采样 30 秒，生成火焰图
py-spy record -o profile.svg --pid 12345 --duration 30 --rate 200
同时在压测端用 Locust 打流量，保证采样期间进程是忙的。火焰图出来后一眼就看到问题：超过 60% 的采样落在 asyncpg 的 fetch 等待上，也就是在等数据库返回。Python 层本身的 CPU 占用很低，说明不是代码逻辑慢，是 IO 等待。
这个结论很关键——它直接排除了"要不要换 Go/Rust 重写"这种错误方向。
3.2 自建分段计时中间件
火焰图只能告诉你"在等数据库"，但一个请求里可能有十几次查询，得知道是哪几次。我加了个轻量中间件，把请求内的 SQL 数量和总耗时打进日志：
# middleware.py
import time
import logging
from starlette.middleware.base import BaseHTTPMiddleware
from sqlalchemy import event
from app.db import engine
logger = logging.getLogger("slowapi")
class ProfilingMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request, call_next):
sql_count = {"n": 0, "t": 0.0}
def before_cursor_execute(conn, cursor, statement, params, context, executemany):
context._query_start = time.perf_counter()
def after_cursor_execute(conn, cursor, statement, params, context, executemany):
cost = time.perf_counter() - context._query_start
sql_count["n"] += 1
sql_count["t"] += cost
event.listen(engine.sync_engine, "before_cursor_execute", before_cursor_execute)
event.listen(engine.sync_engine, "after_cursor_execute", after_cursor_execute)
start = time.perf_counter()
try:
response = await call_next(request)
finally:
event.remove(engine.sync_engine, "before_cursor_execute", before_cursor_execute)
event.remove(engine.sync_engine, "after_cursor_execute", after_cursor_execute)
total = (time.perf_counter() - start) * 1000
if total > 300: # 只记录慢请求
logger.warning(
"SLOW %s %s total=%.1fms sql_count=%d sql_time=%.1fms",
request.method, request.url.path,
total, sql_count["n"], sql_count["t"] * 1000,
)
return response
挂上去跑了一会儿，日志直接把我干沉默了：
SLOW GET /api/v1/products/search total=1842.3ms sql_count=127 sql_time=1710.5ms
SLOW GET /api/v1/products/search total=1903.7ms sql_count=131 sql_time=1755.2ms
一个请求发了 127 次 SQL，SQL 总耗时 1.7 秒，占了总耗时的 93%。 这就是典型的 N+1 查询：先查商品列表（1 次），然后对每个商品循环查库存、查价格、查类目名（每个商品 3 次，一页 40 条就是 120 次）。
到这里，问题定位完成，可以动手了。
四、核心实现：三步优化
4.1 干掉 N+1：改成批量查询 + join
原来的代码大概长这样（简化版）：
# 优化前
products = await session.execute(select(Product).where(...).limit(40))
result = []
for p in products.scalars():
stock = await session.execute(select(Stock).where(Stock.product_id == p.id))
price = await session.execute(select(Price).where(Price.product_id == p.id))
category = await session.execute(select(Category).where(Category.id == p.category_id))
result.append({...})
改法很直接：库存和价格用 IN 批量查一次，类目用 join 一次带出来。如果关系定义得好，直接用 SQLAlchemy 2.0 的 selectinload 最省事：
# 优化后
from sqlalchemy.orm import selectinload
stmt = (
select(Product)
.options(
selectinload(Product.stock),
selectinload(Product.price),
joinedload(Product.category),
)
.where(Product.status == 1)
.order_by(Product.updated_at.desc())
.limit(40)
)
products = (await session.execute(stmt)).scalars().all()
selectinload 会额外发一条 WHERE product_id IN (...) 的查询，而不是 N 条；joinedload 直接 join。改完后 SQL 数量从 127 降到 4 条（count + 主查询 + stock + price）。
4.2 补索引与改分页
SQL 数量降下来了，但单条还是慢。跑 EXPLAIN ANALYZE 看主查询：
Seq Scan on products (cost=0.00..89234.00 rows=1200000 width=...)
Filter: ((status = 1) AND (category_id = 12) AND (price >= 100))
Rows Removed by Filter: 2980000
全表扫描，320 万行扫一遍。而且分页用的是 OFFSET，翻到第 50 页时要先跳过 2000 行再取 40 行，越翻越慢。
两个改动：
（1）加复合索引。 根据查询条件，建了 (status, category_id, updated_at DESC) 的复合索引，价格区间过滤放到索引后面再筛：
CREATE INDEX CONCURRENTLY idx_products_status_cat_updated
ON products (status, category_id, updated_at DESC)
WHERE status = 1;
用了部分索引（WHERE status = 1），因为线上 95% 的查询都只查上架商品，索引体积小了一半。
（2）把 OFFSET 分页改成游标分页。 用 updated_at + id 做游标：
if cursor:
stmt = stmt.where(
tuple_(Product.updated_at, Product.id) str:
raw = json.dumps(params, sort_keys=True, ensure_ascii=False)
return f"{prefix}:{hashlib.md5(raw.encode()).hexdigest()}"
async def cached_search(params: dict, loader):
key = make_key("prodsearch", params)
hit = await redis.get(key)
if hit:
return json.loads(hit)
data = await loader(params)
ttl = 60 + random.randint(0, 15) # 抖动，防雪崩
await redis.set(key, json.dumps(data, ensure_ascii=False), ex=ttl)
return data
另外做了个小的防护：缓存空结果也存（TTL 短一些，10 秒），防止"查不到的关键词"每次都穿透到数据库。
五、踩坑与优化
这部分是我踩过的坑，写出来能帮一个是一个。
坑 1：selectinload 和 limit 的顺序问题。 一开始我在 selectinload 之后加 limit，结果 SQLAlchemy 生成了子查询套 join 的语句，索引失效。后来改成先 limit 主查询、再 selectin，才走对索引。这个要看具体版本行为，2.0.25 上我实测是这样。
坑 2：连接池打满。 优化后单请求 SQL 少了，但并发上来了，4 个 worker 默认连接池（pool_size=5）很快不够用，出现 QueuePool limit overflow。改成 pool_size=20, max_overflow=10, pool_pre_ping=True 后稳定。
坑 3：缓存 key 没考虑分页游标。 第一版 key 只哈希了关键词和类目，忘了游标，导致翻页拿到的是第一页的缓存。后来把 cursor 也纳入 key，并且对第一页单独用长 TTL、后续页用短 TTL。
坑 4：CREATE INDEX 锁表。 直接在业务高峰期跑普通 CREATE INDEX 会锁写，必须用 CONCURRENTLY，虽然慢一点但不阻塞写入。这个一定要在低峰期做，并且监控好。
坑 5：序列化开销。 一开始缓存里存 ORM 对象 pickle，反序列化比查库还慢。改成存纯 dict 的 JSON 后正常了。教训是：缓存里只存 DTO，不存 ORM 实体。
六、效果数据
压测用 Locust，脚本模拟运营后台的真实查询分布（60% 带关键词、30% 带类目、10% 组合条件），持续压 5 分钟：
# locustfile.py
from locust import HttpUser, task, between
import random
KEYWORDS = ["手机", "耳机", "充电器", "数据线", "壳"]
class SearchUser(HttpUser):
wait_time = between(0.1, 0.5)
@task(6)
def search_by_keyword(self):
self.client.get("/api/v1/products/search",
params={"q": random.choice(KEYWORDS), "page_size": 40})
@task(3)
def search_by_category(self):
self.client.get("/api/v1/products/search",
params={"category_id": random.randint(1, 50), "page_size": 40})
@task(1)
def search_combo(self):
self.client.get("/api/v1/products/search",
params={"q": random.choice(KEYWORDS),
"category_id": random.randint(1, 50),
"price_min": 100, "page_size": 40})
前后对比（同样的 4C8G 环境，4 workers）：
| 指标 | 优化前 | 优化后 | 变化 |
|---|---|---|---|
| P50 | 820ms | 45ms | -94.5% |
| P95 | 1840ms | 120ms | -93.5% |
| P99 | 3200ms | 260ms | -91.9% |
| 单机 QPS（P95<500ms） | 210 | 1420 | +576% |
| 单请求 SQL 数 | 127 | 4 | -96.9% |
| 数据库 CPU 峰值 | 85% | 22% | -63pp |
| 缓存命中率 | - | 78% | - |
几个观察：
- QPS 提升主要来自 SQL 数量下降和缓存命中，两者叠加。
- 缓存命中率 78%，比我预期低一点，因为压测脚本里关键词是随机的，真实运营场景重复度更高，线上实测在 85% 左右。
- 数据库 CPU 降到 22%，终于不用天天盯着慢查询告警了。
七、总结
这次调优下来，几点体会：
- 先 profiling 再动手。 py-spy 五分钟就能告诉你瓶颈在哪，比拍脑袋猜一小时靠谱。我见过太多人一上来就"上缓存""换框架"，结果瓶颈在索引上。
- N+1 是 API 性能的头号杀手。 一个请求 127 条 SQL，再怎么优化 Python 代码都没用。ORM 用起来爽，但一定要盯住实际发出的 SQL。
- 索引要按查询模式建。 复合索引的字段顺序、部分索引的条件，都得对着真实查询来，不是越多越好。
- 缓存是放大器不是救命稻草。 数据库没优化好就上缓存，只会把问题延后。我们是先把 SQL 从 127 降到 4、把全表扫描干掉，再加缓存，收益才叠加得起来。
- 压测要贴近真实分布。 如果只压单一关键词，缓存命中率虚高，数据会骗人。
后续还可以做的：把搜索结果预热到本地 LRU、对 count 查询做近似估算、考虑读库分离。但就目前 1400 QPS 的量级，这套方案够用一阵了，不为了优化而优化。
代码都在上面了，profiling 中间件和缓存装饰器可以直接拿去改改用。如果你的接口也在 1 秒以上，先把那个中间件挂上，看看日志里 sql_count 是多少——大概率你会和我一样被吓一跳。
