N+1 查询问题：从一次跨地域接口超时说起

1. 什么是 N+1 查询问题

N+1 查询是 ORM 和分层架构里非常常见的一类性能问题：为了得到一页主表数据，代码先执行 1 次查询拿到 N 条主记录，再在循环里对每条记录各查一次关联数据，一共执行 1 + N 次（甚至更多）数据库查询。

1 次查询：获取审核记录列表（例如 10 条）
N 次查询：对每条记录分别查「主实体」「关联内容 A」「关联内容 B」「统计数量」……
→ 总查询次数 = 1 + 10 × 若干 = 几十次甚至上百次

在单次延迟很低的机房内网环境里，几十次查询可能还能接受；一旦数据库和应用跨地域部署（例如应用与数据库不在同一地域），每次往返 100ms+，总耗时就会轻松突破数秒甚至超时。下面用一次真实优化做说明。

2. 业务背景：UGC 内容审核列表在海外环境超时

部署在国内的运营平台 UGC 内容审核后台有一个审核列表接口：按筛选条件分页拉取待审核/已审核记录，每条记录需要展示：

  • 审核单基本信息
  • 当前生效内容：主实体及其关联的配置、资源等（来自多张表）
  • 最新提交内容：用户最新一次提交的配置、资源及创作者信息
  • 统计信息：该条目的关联记录数量（用于区分新建还是编辑等）
  • 审核人姓名（来自另一套平台自身的用户/权限库）

在海外某区域的生产环境下，该接口经常超时，前端列表迟迟出不来。排查后发现：应用（运营平台）与数据库（海外线上生产数据库）跨地域部署，网络延迟较大，而接口里存在严重的 N+1 查询，把高延迟放大了很多倍。

3. 出问题的实践

3.1 列表接口的原始逻辑（伪代码）

# 1 次查询：拿到本页的审核记录
audit_records, total = await get_audit_list_with_filters(...)

items = []
for audit_record in audit_records:   # 假设 10 条
    # 每条记录都触发多次数据库访问
    effective_content, latest_content = await get_content_versions(
        db, audit_record.entity_id
    )  # 内部：主实体 + 生效配置/资源 + 最新配置/资源 + 创作者 ≈ 6+ 次查询
    
    resource_count = await count_resources_by_entity_id(db, entity_id)   # 1 次
    config_count = await count_configs_by_entity_id(db, entity_id)     # 1 次
    
    item = build_item(audit_record, effective_content, latest_content, ...)
    items.append(item)

# 再批量查审核人姓名（这里本来还可以，但前面已经爆了）
auditor_name_map = await get_auditor_names(user_db, auditor_ids)

也就是说：每一条审核记录都会触发多次跨库/跨表查询。get_content_versions​ 内部还会按 entity_id 再去查主实体、生效配置、生效资源、最新配置、最新资源、创作者等，单条记录就是 6～8 次往返。

3.2 量化：为什么在跨地域环境下会超时

假设：

  • 每页 10 条审核记录
  • 每条记录 8 次数据库往返（保守估计）
  • 应用 ↔ 数据库单次 RTT ≈ 100ms（跨地域常见量级）

则：

  • 总往返次数 ≈ 1（列表） + 10×8 = 81 次
  • 仅网络延迟 ≈ 81 × 100ms ≈ 8.1 秒

再加上 SQL 执行时间、序列化、下游调用等，很容易就超过前端或网关的超时时间，表现为「列表拉取超时」。

4. 解决思路：批量查询 + 并行

思路很简单，把「循环里的单条查询」改成「先批量查，再在内存里组装」：

  1. 先收集本页用到的所有 ID
    例如：所有 entity_id​、所有 auditor_id​、以及由主实体推导出的 config_id​、resource_id 等。

  2. 用少量几次“批量查询”一次性取回

    • 按 entity_id 列表批量查：主实体、最新配置、最新资源、各表统计数量
    • 按 config_id​ / resource_id 列表批量查：生效配置、生效资源
    • 按 creator_id 批量查创作者
    • 审核人姓名：按 auditor_id 在用户库做一次批量查询

  3. 能并行的就并行
    例如：6 个批量查询彼此无依赖，可以用 asyncio.gather 一次发出去，总耗时 ≈ 最慢的那一个，而不是 6 次相加。

  4. 在内存里组装
    遍历本页的 audit_records​，从各个ID → 实体的 map 里取数据，拼成前端需要的结构，不再在循环里访问数据库。

这样，总查询次数从 1 + N×8 降为固定的 7～8 次，且其中 6 次可以并行，高延迟下总耗时主要取决于少数几次往返而不是几十次往返。

5. 业务层批量查询实践

5.1 CRUD 层：提供批量查询

在 CRUD 层为「按 ID 列表查」增加批量接口，避免在 service 里循环调用单条查询。例如：

# 批量查主实体
async def batch_get_entities_by_ids(db, entity_ids: List[int]) -> dict[int, Any]:
    if not entity_ids:
        return {}
    result = await db.execute(
        select(MainEntity).where(MainEntity.id.in_(entity_ids))
    )
    return {row.id: row for row in result.scalars().all()}

# 批量查「每个实体的最新配置」（按 entity_id 分组取 max(id) 再 in 查）
async def batch_get_latest_configs_by_entity_ids(db, entity_ids) -> dict[int, Any]:
    # 子查询：每个 entity_id 对应的最大 config.id
    # 再 in 查这些 id 的完整行
    ...

# 同理：batch_get_latest_resources_by_entity_ids
#       batch_get_configs_by_ids / batch_get_current_resources_by_resource_biz_ids
#       batch_count_resources_by_entity_ids / batch_count_configs_by_entity_ids
#       batch_get_users_by_ids（创作者）

这样，所有“按 ID 查一条”的地方，都改成“按 ID 列表查一批、返回 dict” ，由 service 在内存里做 O(1) 查找。

5.2 Service 层：先批量拉取，再组装

列表接口的 service 逻辑可以改成类似这样（保留你项目里的命名和表结构即可）：

# 1. 拿到本页审核记录（1 次查询）
audit_records, total_count = await get_audit_list_with_filters(...)
entity_ids = list(set(r.entity_id for r in audit_records))

# 2. 批量查本页用到的所有数据（6 个批量查询，并行）
(
    effective_configs_map,
    effective_resources_map,
    latest_configs_map,
    latest_resources_map,
    resource_counts_map,
    config_counts_map,
) = await asyncio.gather(
    batch_get_configs_by_ids(db, effective_config_ids),
    batch_get_current_resources_by_ids(db, effective_resource_ids),
    batch_get_latest_configs_by_entity_ids(db, entity_ids),
    batch_get_latest_resources_by_entity_ids(db, entity_ids),
    batch_count_resources_by_entity_ids(db, entity_ids),
    batch_count_configs_by_entity_ids(db, entity_ids),
)

# 创作者、审核人同样：按 ID 列表各 1 次批量查询
creators_map = await batch_get_users_by_ids(db, creator_ids)
auditor_name_map = await get_auditor_names(user_db, list(auditor_ids))  # 内部改为 in 查询

# 3. 纯内存：遍历 audit_records，从各 map 里取数据组装 item
for audit_record in audit_records:
    entity = entities_map.get(audit_record.entity_id)
    effective_config = effective_configs_map.get(entity.current_config_id)
    # ...
    item = AuditRecordItem(
        ...
        resource_count=resource_counts_map.get(entity_id, 0),
        config_count=config_counts_map.get(entity_id, 0),
    )
    items.append(item)

要点：

  • 循环内不再出现 ​await xxx(db, single_id)​ ，只有「从 dict 里 get」。
  • 与数据库的交互集中在「1 次列表 + 若干次批量 + 1 次审核人」，且批量之间用 gather 并行。

5.3 审核人姓名：用户库也改成批量

如果 get_auditor_names​ 原来是循环里按 auditor_id 逐条查用户，可以改为“一次 in 查询”：

# 用户 DAO 增加批量接口
async def get_users_by_ids(cls, db, user_ids: list[int]) -> list[User]:
    if not user_ids:
        return []
    result = await db.execute(
        select(User).where(User.deleted == 0, User.id.in_(user_ids))
    )
    return list(result.scalars().all())

这样审核人姓名也只会产生 1 次用户库查询，而不是 N 次。

6. 效果对比

指标优化前优化后
每页 10 条时的 DB 往返次数1 + 10×8 ≈ 81 次1 + 6（并行） + 1 ≈ 8 次
仅网络延迟（100ms/次）~8.1s~0.8s（含并行）
接口总耗时易超时明显缩短，可稳定在 2～3 秒内

（典型值）

7. 小结：如何避免和排查 N+1

  1. 写列表/详情接口时
    凡是要“根据本页/本批 ID 去查关联表”的，优先想：能不能先收集 ID，再 in 查一批，最后在内存里组装？ 避免在 for 循环里 await 任何按单 ID 查的接口。

  2. CRUD 层
    除了“按主键查一条”，尽量提供“按 ID 列表查一批”的接口，返回 dict[id -> 实体]​ 或 list，方便 service 做 O(1) 查找。

  3. 无依赖的多次查询
    用 asyncio.gather 并行，总耗时 ≈ max(各查询)，而不是 sum。

  4. 跨地域、高延迟
    延迟越大，N+1 的惩罚越重。在这种环境下，减少往返次数往往比“单次 SQL 再优化”更关键，此时倾向于将查询合并的处理从SQL提到业务代码中来进行合并。

  5. 排查方式

    • 看日志/APM 里该接口的 SQL 条数 和 总耗时；
    • 若“SQL 条数 ≈ 1 + 每页条数 × 常数”，基本就是 N+1；
    • 再在代码里找“循环 + await 查库”的写法，改为批量 + 内存组装。

参考阅读