一、问题背景
去年年底接手了一个内容检索服务，技术栈是 FastAPI + SQLAlchemy + MySQL 8.0 + Redis。接口本身不复杂，GET /api/v1/articles，支持按分类、关键词、分页查询，返回文章列表和作者信息。
上线初期流量不大，没什么感觉。等到日请求量涨到 200 万左右（峰值 QPS 约 600），监控开始报警：P99 延迟 1.8s，P95 也有 700ms，MySQL 的 CPU 使用率长期在 80% 上下浮动，慢查询日志一天能刷出几千条。
产品那边反馈"列表页转圈要两秒"，这不能忍。于是花了两天时间做了一次系统性排查和调优，最终把 P99 压到 120ms。这篇文章把过程完整记下来，包括踩的坑。
先说结论，方便赶时间的同学：
- 主要瓶颈是一条 N+1 查询（列表 20 条，额外发了 20 次作者查询）
- 次要瓶颈是
category_id + created_at缺少联合索引，导致 filesort - 热点分类数据重复查询率高，加 Redis 缓存后数据库压力下降一个数量级
二、环境与版本
先把环境交代清楚，不同版本行为差异挺大的：
- Python 3.11.6
- FastAPI 0.110.0
- Uvicorn 0.27.1（
--workers 4，跑在 4C8G 的容器里） - SQLAlchemy 2.0.27（async 模式，aiomysql 0.2.0 驱动）
- MySQL 8.0.35，InnoDB，buffer pool 4G
- Redis 7.2.4
- 压测工具：locust 2.20.1
- profiling：py-spy 0.3.14、FastAPI 自定义 Middleware 计时
原始接口的核心代码大概长这样（简化过）：
@router.get("/articles")
async def list_articles(
category_id: int | None = None,
keyword: str | None = None,
page: int = 1,
size: int = 20,
db: AsyncSession = Depends(get_db),
):
stmt = select(Article).where(Article.status == 1)
if category_id:
stmt = stmt.where(Article.category_id == category_id)
if keyword:
stmt = stmt.where(Article.title.like(f"%{keyword}%"))
stmt = stmt.order_by(Article.created_at.desc()).offset((page - 1) * size).limit(size)
result = await db.execute(stmt)
articles = result.scalars().all()
items = []
for a in articles:
author = await db.get(Author, a.author_id) # 就是这里，N+1
items.append({
"id": a.id,
"title": a.title,
"created_at": a.created_at,
"author": {"id": author.id, "name": author.name},
})
return {"items": items, "page": page, "size": size}
这段代码看着人畜无害，问题全藏在细节里。下面一步步拆。
三、方案设计：先定位，再动手
我的原则是：没有 profiling 数据之前不猜瓶颈。所以分三步走。
第一步，用 py-spy 抓火焰图。 py-spy 的好处是 attach 到运行中的进程，不用改代码、不用重启。生产容器里直接：
py-spy record -o profile.svg --pid 12345 --duration 60 --rate 200
同时开 locust 打流量，抓 60 秒。火焰图上 aiomysql 的 read_packet 和 SQLAlchemy 的 _execute_context 占了将近 60% 的宽度，基本实锤是数据库交互的问题。
第二步，加一个轻量 Middleware 记录 SQL 次数。 py-spy 只能告诉你"慢在数据库"，但不知道慢在哪条 SQL。我在测试环境挂了个中间件，通过 SQLAlchemy 的 event 钩子统计单次请求内的 SQL 条数：
from sqlalchemy import event
from starlette.middleware.base import BaseHTTPMiddleware
class SQLCountMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request, call_next):
counter = {"n": 0, "sql": []}
def before_cursor_execute(conn, cursor, statement, params, context, executemany):
counter["n"] += 1
counter["sql"].append(statement[:120])
engine = request.app.state.engine
event.listen(engine.sync_engine, "before_cursor_execute", before_cursor_execute)
try:
response = await call_next(request)
finally:
event.remove(engine.sync_engine, "before_cursor_execute", before_cursor_execute)
if counter["n"] > 10:
print(f"[SLOW-SQL] {request.url.path} count={counter['n']}")
for s in counter["sql"]:
print(" ", s)
return response
跑一次列表请求，输出直接暴露问题：一次请求发了 21 条 SQL——1 条列表查询 + 20 条 SELECT * FROM author WHERE id = ?。典型的 N+1。
第三步，开 MySQL 慢查询日志看单条 SQL 的执行计划。 long_query_time 设成 0.1，用 EXPLAIN 看列表查询：
type=ref, key=idx_status, rows=85000, Extra=Using filesort
扫了 8.5 万行还要 filesort，因为 ORDER BY created_at 没有走索引。这就是第二个瓶颈。
定位清楚了，方案就三件事：消除 N+1、补索引、加缓存。
四、核心实现
4.1 用 selectinload 消除 N+1
SQLAlchemy 2.0 的 async 模式下，selectinload 会额外发一条 WHERE id IN (...) 把关联对象批量捞回来，20 条列表变成 2 条 SQL。前提是模型上配好 relationship：
class Article(Base):
__tablename__ = "article"
id: Mapped[int] = mapped_column(primary_key=True)
title: Mapped[str] = mapped_column(String(200))
category_id: Mapped[int] = mapped_column(index=True)
author_id: Mapped[int] = mapped_column(ForeignKey("author.id"))
status: Mapped[int] = mapped_column(default=1)
created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())
author: Mapped["Author"] = relationship(lazy="raise") # 关键：禁止隐式懒加载
lazy="raise" 这行很重要。默认的 lazy="select" 在 async 下会直接抛 MissingGreenlet，但如果有人切回同步 session，隐式懒加载会悄悄把 N+1 带回来。设成 raise 之后，忘写 selectinload 会立刻报错，属于"把 bug 变成异常"的防御性写法。
改写后的查询：
from sqlalchemy.orm import selectinload
stmt = (
select(Article)
.options(selectinload(Article.author))
.where(Article.status == 1)
)
if category_id:
stmt = stmt.where(Article.category_id == category_id)
if keyword:
stmt = stmt.where(Article.title.like(f"%{keyword}%"))
stmt = stmt.order_by(Article.created_at.desc()).offset((page - 1) * size).limit(size)
result = await db.execute(stmt)
articles = result.scalars().all()
items = [
{
"id": a.id,
"title": a.title,
"created_at": a.created_at,
"author": {"id": a.author.id, "name": a.author.name},
}
for a in articles
]
一次请求 SQL 从 21 条降到 2 条。这一步单独做完，P99 就从 1.8s 掉到大概 600ms。
4.2 补联合索引
列表页最常见的查询是"某分类下按时间倒序"，所以建联合索引，注意顺序：等值条件在前，排序字段在后。
ALTER TABLE article
ADD INDEX idx_cat_status_created (category_id, status, created_at DESC),
ALGORITHM=INPLACE, LOCK=NONE;
ALGORITHM=INPLACE, LOCK=NONE 是为了在线加索引不锁表，8.0 上一般没问题，但生产加索引前还是建议在从库先验证一遍耗时。这张表当时 1200 万行，加索引用了 40 秒左右。
加完再看 EXPLAIN：type=ref, rows=1200, Extra=Using index condition，filesort 消失。单条列表查询从平均 180ms 降到 8ms。
这里有个坑后面细说：keyword 的 LIKE '%xx%' 是走不了这个索引的，前导通配符天然失效，这个后面单独处理。
4.3 Redis 缓存热点分类
索引和 N+1 都修完，P99 大概 200ms 出头。但监控显示 QPS 峰值时数据库还是有压力，因为首页和几个热门分类的请求占了总流量的 70% 以上，同样的数据被反复查。
上缓存。策略很简单：只缓存第一页、无关键词的列表请求，key 用 article:list:{category_id}:p1:s20，TTL 60 秒，接受一分钟的最终一致。带关键词的搜索不缓存，因为组合太多、命中率低，缓存反而制造垃圾。
import json
import redis.asyncio as aioredis
redis = aioredis.from_url("redis://127.0.0.1:6379/0", decode_responses=True)
async def get_article_list(db, category_id, page, size, keyword):
cacheable = keyword is None and page == 1
cache_key = f"article:list:{category_id or 0}:p{page}:s{size}"
if cacheable:
cached = await redis.get(cache_key)
if cached:
return json.loads(cached)
# ... 执行 4.1 里的查询，组装 items ...
payload = {"items": items, "page": page, "size": size}
if cacheable:
await redis.set(cache_key, json.dumps(payload, default=str), ex=60)
return payload
为了防缓存击穿，热门 key 过期瞬间可能有几十个请求同时打到数据库。加了个简单的分布式锁：抢到锁的请求去查库回填，没抢到的 sleep 50ms 后重读缓存。
lock_key = f"lock:{cache_key}"
if await redis.set(lock_key, "1", nx=True, ex=5):
try:
# 查库、回填缓存
...
finally:
await redis.delete(lock_key)
else:
await asyncio.sleep(0.05)
cached = await redis.get(cache_key)
if cached:
return json.loads(cached)
# 兜底走查库
五、踩坑与优化
坑一：LIKE '%keyword%' 打回原形。 索引和缓存都只优化了"无关键词"路径，带关键词的请求还是全表扫。1200 万行的 LIKE '%xx%' 单次要 1.2s 以上。短期内上了 MySQL 8.0 的全文索引（FULLTEXT + MATCH ... AGAINST）覆盖标题搜索，把这类请求从 1.2s 压到 80ms 左右。长期方案是上 ES，但那是另一个话题了。全文索引对中文需要配 ngram parser，这个配置本身也有坑，ngram_token_size=2 比较适合中文短词。
坑二：lazy="raise" 上线后炸了几个老接口。 有些后台管理接口之前一直靠隐式懒加载跑着，改成 raise 之后直接 500。这个属于预期内的"暴露问题"，逐个补 selectinload 就好，但提醒一下：改 lazy 策略要全量回归，别只测主链路。
坑三：缓存 TTL 60 秒导致编辑后看不到更新。 运营改了文章标题，列表页一分钟内还是旧的，被投诉了。后来在写接口里加了一步主动删缓存：文章更新时 DEL article:list:{category_id}:p1:s20。简单粗暴但有效。
坑四：连接池配置。 压测时出现过 TimeoutError: QueuePool limit of size 20 overflow 10 reached。4 个 worker，每个默认 pool_size 5、max_overflow 10，峰值并发一上来就不够。调到 pool_size=20, max_overflow=30, pool_recycle=1800 后稳定。注意别把 pool 开太大把 MySQL 连接数打满，要算总账：worker 数 × pool_size 不能超过 MySQL 的 max_connections 减去其他服务的占用。
六、效果数据
压测用 locust，4 个 worker 的 Uvicorn，混合场景：70% 无关键词列表（可缓存）、30% 带关键词搜索。压测机 8C16G，持续 10 分钟。
| 指标 | 优化前 | 优化后 |
|---|---|---|
| P50 | 420ms | 45ms |
| P95 | 700ms | 95ms |
| P99 | 1800ms | 120ms |
| 单请求 SQL 条数 | 21 | 2（缓存命中时 0） |
| 峰值 QPS（P99<200ms） | 约 600 | 约 2600 |
| MySQL 峰值 QPS | 3200 | 400 |
| MySQL CPU | 82% | 18% |
| 缓存命中率 | - | 76% |
数据库 QPS 从 3200 降到 400，这个降幅主要来自缓存，N+1 消除贡献了 SQL 条数的下降。真正的收益是"数据库不再是瓶颈"，后面流量再涨两倍也不用急着分库分表。
七、总结
回头看，这次调优没什么高深技巧，就是老老实实按顺序来：
- 先 profiling 再动手。py-spy 火焰图 + SQL 计数中间件，半小时就能定位到 N+1，比拍脑袋猜快得多。
- N+1 是列表接口的头号杀手。async 模式下用
selectinload，配合lazy="raise"防止回潮。 - 联合索引注意字段顺序：等值列在前、排序列在后，加索引用
ALGORITHM=INPLACE, LOCK=NONE。 - 缓存只缓存值得缓存的。第一页热点数据 TTL 60 秒，命中率 76%，比什么都缓存效果好得多。
- 每个优化都要有压测数据佐证，不然你不知道是自己变快了还是流量变小了。
最后留个待办：LIKE '%xx%' 的全文索引方案目前只能算止血，等文章量再涨一波，搜索这块还是得上 ES。到时候再写一篇。
