ORM 深水区：连接池与 N+1 排查

第 17 章 · ORM 深水区：连接池、读写分离与 N+1 排查

本章目标：深入 SQLAlchemy Engine 连接池配置（pool_size、pool_pre_ping、pool_recycle）；理解读写分离（主库写、从库读）及 SQLALCHEMY_BINDS 示例；用 joinedload / selectinload 解决 N+1；掌握子查询、exists、聚合与窗口函数简介；配置 SQL 日志（对标 Django Debug Toolbar）；建立慢查询排查 checklist。

学时建议：6～7 小时（含 2 小时性能实验）

前置：本模块 ch05 ORM、ch09 日志、ch11 生产部署；ch16 RBAC 列表查询可作为优化对象。

17.1 性能问题从哪来

api-demo 商品量上涨后的典型症状：

现象	可能原因
首请求慢、后续变快	连接池冷启动
凌晨后 API 报错	连接被 `wait_timeout` 断开
列表 10 条打 11 次 SQL	N+1 懒加载
报表接口超时	缺索引、全表扫描
写后立刻读不到	主从复制延迟

 Flask Worker              PostgreSQL / MySQL
┌────────────┐            ┌────────────────────┐
│ SQLAlchemy │── pool ──►│ db-primary.example │ 写
└────────────┘            └─────────┬──────────┘
      │ 读查询（选修）                │ 复制
      └──────────────────────────►┌────────────────────┐
                                  │ db-replica.example │ 读
                                  └────────────────────┘

数据库主机均为虚构域名；本地可用 SQLite，生产示例用 PostgreSQL 语法。

17.2 Engine 与连接池核心参数

Flask-SQLAlchemy 在 db.init_app 时根据 URI 创建 Engine。

api_demo/config.py（ProductionConfig）：

class ProductionConfig(Config):
    SQLALCHEMY_DATABASE_URI = os.environ.get(
        "DATABASE_URL",
        "postgresql+psycopg2://api_demo:***@db-primary.example.com:5432/api_demo",
    )
    SQLALCHEMY_ENGINE_OPTIONS = {
        "pool_size": 10,
        "max_overflow": 20,
        "pool_timeout": 30,
        "pool_recycle": 1800,
        "pool_pre_ping": True,
    }

参数	含义	建议
`pool_size`	常驻连接数	每 Worker 5～20
`max_overflow`	峰值额外连接	突发缓冲，不宜过大
`pool_pre_ping`	取出前 `SELECT 1`	生产建议 True
`pool_recycle`	连接存活秒数	MySQL 设 < `wait_timeout`
`pool_timeout`	等待连接秒数	防线程无限阻塞

连接数估算：总连接 ≈ Gunicorn workers × (pool_size + overflow) + Celery workers × 池大小。示例 4×15+20=80，数据库 max_connections 须留余量。开发环境 SQLite 无网络池，调优在 staging / 生产验证。

17.3 读写分离：主库写、从库读

操作	目标库	说明
INSERT/UPDATE/DELETE	Primary	唯一写入点
SELECT（可接受延迟）	Replica	分担读压
事务内读 / 写后立刻读	Primary	「读己之写」

Binds 配置：

class ProductionConfig(Config):
    SQLALCHEMY_DATABASE_URI = os.environ.get(
        "DATABASE_PRIMARY_URL",
        "postgresql+psycopg2://api_demo:***@db-primary.example.com:5432/api_demo",
    )
    SQLALCHEMY_BINDS = {
        "replica": os.environ.get(
            "DATABASE_REPLICA_URL",
            "postgresql+psycopg2://api_demo:***@db-replica.example.com:5432/api_demo",
        ),
    }

显式路由（优于给写模型设 __bind_key__ = "replica"）：

from flask import g

def get_read_session():
    if "read_session" not in g:
        g.read_session = db.create_session(bind=db.engines["replica"])()
    return g.read_session

@app.teardown_appcontext
def shutdown_read_session(exception=None):
    sess = g.pop("read_session", None)
    if sess:
        sess.close()

读列表走从库，写仍用 db.session（主库）：

@api_bp.route("/products")
def product_list():
    return get_read_session().query(Product).filter_by(is_published=True).paginate(...)

# 写
db.session.add(Product(name="..."))
db.session.commit()

复制延迟 50～500ms：写后详情读主库，或前端乐观更新。

17.4 N+1 问题：识别与修复

products = Product.query.limit(10).all()   # 1 次
for p in products:
    print(p.category.name)                 # 每条 +1 → 共 11 次

joinedload（多对一/一对一，一次 JOIN）：

from sqlalchemy.orm import joinedload

products = (
    Product.query
    .options(joinedload(Product.category))
    .filter_by(is_published=True).limit(10).all()
)

selectinload（一对多/多对多，两次 IN 查询）：

from sqlalchemy.orm import selectinload

categories = Category.query.options(selectinload(Category.products)).all()

场景	策略
商品列表 + 分类	`joinedload(Product.category)`
分类 + 其下商品	`selectinload(Category.products)`
商品 + 创建者（ch16）	`joinedload(Product.created_by)`
仅要少数字段	`load_only(Product.id, Product.name)`

一对多慎用 joinedload，易产生行膨胀；组合示例：

Product.query.options(
    joinedload(Product.category),
    joinedload(Product.created_by),
).order_by(Product.id.desc()).paginate(page=page, per_page=20)

17.5 复杂查询进阶

子查询——各分类商品数：

from sqlalchemy import func

subq = (
    db.session.query(Product.category_id, func.count(Product.id).label("cnt"))
    .group_by(Product.category_id).subquery()
)
rows = db.session.query(Category.name, subq.c.cnt).outerjoin(
    subq, Category.id == subq.c.category_id
).all()