第 17 章 · 异步性能、连接池与 N+1 排查
本章目标：理解 FastAPI async def 何时真正获益、何时应使用同步 def；配置 SQLAlchemy 2.0 异步引擎 pool_size / max_overflow / pool_pre_ping；运用 selectinload / joinedload 消除 N+1；使用 SQLAlchemy echo、logging、EXPLAIN 定位慢查询；入门 压测概念（wrk / locust / k6）与 RED 指标基线；建立 N+1 回归测试。
学时建议：6～7 小时（含 2 小时性能实验）
前置：本模块 ch05 异步 SQLAlchemy、ch08 中间件与日志、ch12 svc-demo MVP；ch16 RBAC 列表查询可作为优化对象。
17.1 性能问题从哪来
svc-demo 商品量上涨后的典型症状：
| 现象 | 可能原因 |
|---|---|
| 并发高时响应变慢 | 连接池耗尽、事件循环阻塞 |
| 凌晨后 API 报错 | 连接被 idle_timeout 断开 |
| 列表 10 条打 11 次 SQL | N+1 懒加载 |
| 报表接口超时 | 缺索引、全表扫描 |
| CPU 100% 但 QPS 低 | async 路由内调用阻塞库 |
Client Uvicorn (async) PostgreSQL
┌────────┐ ┌─────────────────┐ ┌──────────────────┐
│ 浏览器 │────►│ FastAPI + ORM │───►│ db.example.com │
└────────┘ │ asyncpg pool │ │ pool_size=10 │
└─────────────────┘ └──────────────────┘数据库主机为虚构域名 db.example.com；本地可用 SQLite 学 ORM，连接池在 staging 用 PostgreSQL 验证。
17.2 async def 何时用
FastAPI 运行在 ASGI 上：async def 路由在事件循环中协作式调度；def 路由由线程池执行，适合阻塞代码。
| 场景 | 推荐 | 原因 |
|---|---|---|
await session.execute(...) | async def | 真异步 IO |
await httpx.AsyncClient.get | async def | 非阻塞 HTTP |
time.sleep(5) 在 async 路由 | 禁止 | 阻塞整个事件循环 |
requests.get 在 async 路由 | 避免 | 改用 httpx 或 def + 线程池 |
| CPU 密集（图像处理） | def 或 run_in_executor | 不占事件循环 |
| 同步 ORM（无 async 驱动） | def | 官方推荐阻塞 ORM 用同步路由 |
黄金法则：async 路由内只能 await 异步库；混用阻塞调用会拖垮全部并发。
# ❌ 错误：async 路由内阻塞
@router.get("/bad")
async def bad():
import requests
return requests.get("https://api.example.com/ping").json()
# ✅ 正确
@router.get("/good")
async def good():
async with httpx.AsyncClient() as client:
r = await client.get("https://api.example.com/ping")
return r.json()run_in_executor 包装遗留同步代码（尽量少用）：
import asyncio
from functools import partial
@router.post("/legacy")
async def call_legacy():
loop = asyncio.get_running_loop()
result = await loop.run_in_executor(None, partial(sync_heavy_fn, arg=1))
return {"ok": result}17.3 连接池：pool_size 与 max_overflow
SQLAlchemy 2.0 异步引擎（svc_demo/db/session.py）：
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
DATABASE_URL = "postgresql+asyncpg://svc_demo:***@db.example.com:5432/svc_demo"
engine = create_async_engine(
DATABASE_URL,
echo=False,
pool_size=10, # 常驻连接数
max_overflow=20, # 峰值额外连接 = 10 + 20 = 30
pool_timeout=30, # 等待空闲连接超时（秒）
pool_pre_ping=True, # 取用前 SELECT 1，避免 stale 连接
pool_recycle=1800, # 30 分钟回收，应对 idle_timeout
)
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)| 参数 | 含义 | 建议 |
|---|---|---|
pool_size | 池内常驻连接 | 按并发与 worker 数估算 |
max_overflow | 超出 pool_size 的临时连接 | 生产设上限，防打爆 DB |
pool_pre_ping | 连接健康检查 | 生产建议 True |
pool_recycle | 定期重建连接 | 小于 DB idle_timeout |
连接数估算：
总连接 ≈ Uvicorn workers × (pool_size + 实际 overflow) + Celery workers × 连接数示例：4 workers × 10 pool_size ≈ 40，加 overflow 峰值需 < PostgreSQL max_connections。
pgBouncer（概念）
多进程部署时，可在应用前加 pgBouncer：应用连 6432，PgBouncer 用少量连接连真实库 5432。
| 模式 | 说明 |
|---|---|
| session | 会话级，兼容最好 |
| transaction | 事务结束归还，FastAPI 常用 |
| statement | 最激进，部分 ORM 特性受限 |
连 PgBouncer 时可适当减小 pool_size，由代理统一池化。
17.4 N+1 问题与 selectinload
N+1：查询 N 条主记录后，访问关联字段再触发 N 次额外查询。
# ❌ N+1：每条 product 访问 category 再打一次 SQL
products = (await db.scalars(select(Product).limit(10))).all()
for p in products:
print(p.category.name) # 懒加载触发方案 A：selectinload（推荐，1+N 变 2 条 SQL）
from sqlalchemy.orm import selectinload
stmt = (
select(Product)
.options(selectinload(Product.category))
.limit(10)
)
products = (await db.scalars(stmt)).all()方案 B：joinedload（单条 JOIN，注意行膨胀）
from sqlalchemy.orm import joinedload
stmt = select(Product).options(joinedload(Product.category)).limit(10)| 策略 | 适用 |
|---|---|
selectinload | 一对多、多对多，列表页 |
joinedload | 多对一，详情页 |
subqueryload | 特殊报表（少用） |
显式 join + contains_eager | 复杂过滤 |
多层关联：
stmt = select(Order).options(
selectinload(Order.items).selectinload(OrderItem.product),
)17.5 查询优化工具箱
only / defer 减少列
stmt = select(Product).options(load_only(Product.id, Product.name, Product.price))分页与索引
stmt = select(Product).order_by(Product.id).offset(skip).limit(limit)确保 WHERE created_by_id = ?、ORDER BY id 有索引（ch06 Alembic 迁移可加）。
EXPLAIN ANALYZE（概念）
EXPLAIN ANALYZE
SELECT * FROM products WHERE created_by_id = 42 ORDER BY id DESC LIMIT 20;关注 Seq Scan（全表扫描）→ 考虑索引；Nested Loop 行数爆炸 → 检查 JOIN 条件。
SQL 计数回归测试
tests/test_query_count.py：
import pytest
from sqlalchemy import event
@pytest.mark.asyncio
async def test_list_products_query_count(db, client, admin_token):
count = {"n": 0}
def before_cursor_execute(conn, cursor, statement, parameters, context, executemany):
count["n"] += 1
event.listen(db.sync_session.bind.sync_engine, "before_cursor_execute", before_cursor_execute)
r = await client.get(
"/api/v1/products",
headers={"Authorization": f"Bearer {admin_token}"},
)
assert r.status_code == 200
assert count["n"] <= 3 # 列表 + count + selectinload 批次