一、问题背景
事情是这样的。我们有个内部数据平台，前端有个"订单列表"页面，接口是 FastAPI 写的，路径 /api/v1/orders。上线初期数据量小，没人抱怨。三个月后订单表到了 480 万行，运营开始反馈"这个页面转圈要好几秒"。
我去看了下监控（Prometheus + Grafana，接口层用 Starlette 中间件打点），数据挺难看：
- P50 延迟：820ms
- P99 延迟：2340ms
- QPS 峰值：42
- 容器 CPU 使用率：稳定在 28%~33%
CPU 没打满，延迟却这么高，第一反应就是 I/O 等待。这个接口逻辑不复杂：查订单主表、关联查用户信息、关联查订单明细、再查一次商品信息，最后组装返回。典型的多表关联查询场景，如果每张表都单独发一次 SQL，往返次数就很可观了。
先贴一下当时的简化版代码，方便后面讨论：
# app/api/orders.py (优化前)
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from app.db import get_db
from app import models, schemas
router = APIRouter()
@router.get("/api/v1/orders", response_model=list[schemas.OrderOut])
def list_orders(page: int = 1, size: int = 20, db: Session = Depends(get_db)):
orders = (
db.query(models.Order)
.order_by(models.Order.created_at.desc())
.offset((page - 1) * size)
.limit(size)
.all()
)
result = []
for o in orders:
user = db.query(models.User).filter(models.User.id == o.user_id).first()
items = db.query(models.OrderItem).filter(models.OrderItem.order_id == o.id).all()
item_list = []
for it in items:
product = db.query(models.Product).filter(models.Product.id == it.product_id).first()
item_list.append({"product": product.name, "qty": it.qty})
result.append({
"id": o.id,
"user": user.name if user else None,
"items": item_list,
})
return result
一页 20 条订单，每条要查 1 次用户、1 次明细、明细里每个商品再查 1 次。假设平均每单 3 个商品，那就是 20 × (1 + 1 + 3) = 100 次 SQL。这就是经典的 N+1 问题，而且是最严重的那种嵌套 N+1。
二、环境与版本
先把环境交代清楚，避免读者对不上号：
| 组件 | 版本 |
|---|---|
| Python | 3.11.6 |
| FastAPI | 0.109.2 |
| Starlette | 0.36.3 |
| SQLAlchemy | 2.0.25（ORM 模式） |
| Uvicorn | 0.27.0（4 workers，gunicorn 管理） |
| PostgreSQL | 15.5 |
| Redis | 7.2.3 |
| py-spy | 0.3.14 |
| locust | 2.20.0 |
部署在 4C8G 的容器里，数据库是独立实例（8C16G），网络 RTT 约 0.4ms。
三、方案设计：先定位，再动手
我的原则是：没有数据支撑的优化都是猜。所以整个流程分三步走。
第一步，确认瓶颈在哪一层。 用 py-spy 对运行中的进程采样，看时间花在 Python 代码还是等 I/O。py-spy 的好处是不用改代码、不用重启，直接 attach 到 PID：
# 采样 30 秒，生成火焰图
py-spy record -o profile.svg --pid $(pgrep -f "uvicorn app.main:app") --duration 30
# 也可以直接 top 模式看实时热点
py-spy top --pid $(pgrep -f "uvicorn app.main:app")
火焰图出来之后很直观：超过 70% 的采样落在 psycopg2 的 execute 和 wait 上，Python 层的业务代码占比不到 5%。结论明确——时间全花在数据库往返上。
第二步，量化 SQL 次数。 光看火焰图还不够，得知道到底发了多少条 SQL。SQLAlchemy 提供了事件钩子，挂一个计数器就行，比开 echo=True 干净得多：
# app/db.py 里临时加的性能探针
from sqlalchemy import event
from sqlalchemy.engine import Engine
import logging
logger = logging.getLogger("sql_counter")
_sql_count = {"n": 0}
@event.listens_for(Engine, "before_cursor_execute")
def _count_sql(conn, cursor, statement, parameters, context, executemany):
_sql_count["n"] += 1
logger.debug("SQL: %s", statement[:200])
def reset_sql_count():
_sql_count["n"] = 0
def get_sql_count():
return _sql_count["n"]
然后在接口里临时打印，压测一轮，发现单次请求平均执行 103 条 SQL，和估算完全吻合。
第三步，逐条 SQL 分析。 把慢 SQL 抓出来跑 EXPLAIN (ANALYZE, BUFFERS)，发现两个问题：
orders表按created_at DESC排序分页，created_at上没索引，每次都是全表 Seq Scan + Sort，480 万行排序一次 400ms 起。order_items.order_id和products.id上虽然有索引，但查询次数太多，单条再快也架不住 100 次往返。
方案就清晰了，三层改造：
- 数据库层：加复合索引，消灭排序和全表扫描；
- ORM 层：用
selectinload预加载关联，把 N+1 压成固定 3 条 SQL； - 缓存层：热点分页数据进 Redis，TTL 60 秒，扛住重复请求。
四、核心实现
4.1 索引优化
针对排序分页，建复合索引。注意 ORDER BY created_at DESC 配合 OFFSET/LIMIT，索引方向要和排序方向一致：
-- 覆盖排序 + 分页的复合索引
CREATE INDEX CONCURRENTLY idx_orders_created_at_id
ON orders (created_at DESC, id DESC);
-- 明细表按订单查，顺便把常用字段带上做覆盖索引
CREATE INDEX CONCURRENTLY idx_order_items_order_id
ON order_items (order_id) INCLUDE (product_id, qty);
用 CONCURRENTLY 是必须的，480 万行的表直接 CREATE INDEX 会锁表，线上不能这么干（这个坑我后面还会提）。
改完之后单看那条分页 SQL，从 420ms 降到 3ms 左右，EXPLAIN 显示走了 Index Scan。
4.2 消灭 N+1
SQLAlchemy 2.0 里预加载有几种方式，joinedload 是 JOIN 一次查完，selectinload 是主查询 + 一次 IN 查询。对于一对多（订单对明细）场景，selectinload 更合适，避免 JOIN 导致的笛卡尔积膨胀。改写后：
# app/api/orders.py (优化后)
from fastapi import APIRouter, Depends, Query
from sqlalchemy.orm import Session, selectinload
from sqlalchemy import select
from app.db import get_db
from app import models, schemas
from app.cache import cache_get, cache_set
router = APIRouter()
@router.get("/api/v1/orders", response_model=list[schemas.OrderOut])
def list_orders(
page: int = Query(1, ge=1),
size: int = Query(20, ge=1, le=100),
db: Session = Depends(get_db),
):
cache_key = f"orders:list:{page}:{size}"
if (cached := cache_get(cache_key)) is not None:
return cached
stmt = (
select(models.Order)
.options(
selectinload(models.Order.user),
selectinload(models.Order.items).selectinload(models.OrderItem.product),
)
.order_by(models.Order.created_at.desc(), models.Order.id.desc())
.offset((page - 1) * size)
.limit(size)
)
orders = db.execute(stmt).scalars().all()
result = [
{
"id": o.id,
"user": o.user.name if o.user else None,
"items": [
{"product": it.product.name, "qty": it.qty}
for it in o.items
],
}
for o in orders
]
cache_set(cache_key, result, ttl=60)
return result
这里有个细节：selectinload(models.Order.items).selectinload(models.OrderItem.product) 是链式预加载，第二层也会被合并成一次 IN 查询。最终 SQL 数量从 103 条降到 3 条（订单主查询 + 用户 IN 查询 + 明细 IN 查询 + 商品 IN 查询，实际是 4 条，商品那层因为明细已加载所以合并了）。
4.3 Redis 缓存
缓存层用 redis-py，注意 JSON 序列化和连接池配置：
# app/cache.py
import json
import redis
from app.config import settings
_pool = redis.ConnectionPool.from_url(
settings.REDIS_URL,
max_connections=50,
socket_timeout=0.5,
socket_connect_timeout=0.5,
decode_responses=True,
)
_client = redis.Redis(connection_pool=_pool)
def cache_get(key: str):
try:
raw = _client.get(key)
return json.loads(raw) if raw else None
except redis.RedisError:
return None # 缓存故障降级，不影响主流程
def cache_set(key: str, value, ttl: int = 60):
try:
_client.setex(key, ttl, json.dumps(value, ensure_ascii=False, default=str))
except redis.RedisError:
pass
socket_timeout=0.5 是关键配置。缓存是加速手段，不能反过来成为故障点，超时短一点、异常吞掉，降级到数据库就行。
4.4 压测脚本
用 locust 写个简单的压测，模拟 200 并发持续 3 分钟：
# locustfile.py
from locust import HttpUser, task, between
class OrderUser(HttpUser):
wait_time = between(0.1, 0.5)
@task(10)
def list_orders(self):
self.client.get("/api/v1/orders?page=1&size=20", name="/orders list")
@task(1)
def list_orders_deep(self):
self.client.get("/api/v1/orders?page=50&size=20", name="/orders deep page")
跑的时候记得关掉 SQL 日志，不然日志 I/O 会污染结果。
五、踩坑与优化
坑一：CREATE INDEX 锁表。 第一次我图省事直接 CREATE INDEX，结果 DDL 拿了 SHARE 锁，写入全部阻塞，告警响了。后来改成 CONCURRENTLY，建索引期间不阻塞读写，代价是耗时更长（480 万行建了约 90 秒）。生产环境切记。
坑二：OFFSET 深分页。 索引加完后，page=1 很快，但 page=50 依然慢。因为 OFFSET 980 还是要扫描并丢弃前 980 行。压测里我特意加了深分页任务，P99 一度被它拖到 600ms。最终方案是缓存兜底 + 限制最大页数（业务上没人翻到 50 页之后），真要支持深分页得上游标分页（keyset pagination），这次没做。
坑三：缓存击穿。 TTL 60 秒到期瞬间，200 并发全打到数据库，QPS 瞬间飙到 800，数据库连接池直接打满。加了个简单的互斥锁（Redis SET NX）让一个请求回源，其余等待：
def cache_get_or_lock(key: str):
if (v := cache_get(key)) is not None:
return v, False
locked = _client.set(f"{key}:lock", "1", nx=True, ex=5)
return None, bool(locked)
坑四：selectinload 的 IN 参数上限。 PostgreSQL 对 IN 列表长度没硬限制，但参数太多时执行计划会退化。一页 20 条订单还好，后来有人把 size 改成 500，IN 里塞了 1500 个 ID，查询反而变慢。所以我在接口上加了 le=100 的约束。
坑五：序列化开销。 优化到 200ms 左右时，火焰图里 json.dumps 开始显眼。缓存层我用的 json.dumps，大列表序列化要 15ms 左右。后来换成 orjson，降到 4ms。这个属于锦上添花，不是主要矛盾。
六、效果数据
压测条件统一：locust 200 并发，持续 3 分钟，4 worker uvicorn，同一台压测机（避免网络干扰）。
| 指标 | 优化前 | 优化后 | 变化 |
|---|---|---|---|
| P50 延迟 | 820ms | 45ms | -94.5% |
| P95 延迟 | 1680ms | 120ms | -92.9% |
| P99 延迟 | 2340ms | 180ms | -92.3% |
| QPS | 42 | 610 | +1352% |
| 单请求 SQL 数 | 103 | 4（未命中缓存）/ 0（命中） | -96% |
| 容器 CPU | 31% | 58% | 上升（合理，之前都在等 I/O） |
| PG 连接数峰值 | 98 | 22 | -77% |
逐层拆解一下贡献：
- 只加索引：P99 从 2340ms → 1900ms（排序变快，但 SQL 次数没变）
- 加索引 + selectinload：P99 → 320ms（主要收益在这里）
- 再加 Redis 缓存：P99 → 180ms（命中率约 78%，因为压测请求集中在 page=1）
缓存命中率 78% 不算高，因为压测脚本随机打不同页码。真实线上 page=1 占比超过 90%，命中率会更高。
七、总结
这次优化没什么黑科技，就是老老实实按"定位 → 量化 → 改造 → 验证"走了一遍。几点体会：
- CPU 不高但延迟高，优先怀疑 I/O 和 N+1，别一上来就加机器。
- py-spy 是排查线上性能问题的第一工具，attach 即用，对生产零侵入，比改代码加日志快得多。
- SQLAlchemy 的 N+1 是隐形杀手，写代码时觉得"一行一个 query 很清晰"，一页 20 条就是 100 次往返。
selectinload应该是列表接口的默认姿势。 - 缓存要有降级和防击穿设计，否则加速手段会变成故障源。
- 深分页别用 OFFSET，业务允许就限制页数，不允许就上游标分页。
最后说一句，优化前后我都留了压测脚本和 SQL 计数器在仓库里，下次谁再改这个接口，跑一遍就知道有没有退化。性能这东西，没有回归测试就是玄学。
