企业官网建设流程全解析

MongoDB Router Role API 深度解析：CollectionRouter 路由框架与 Scatter-Gather 命令分发机制

【免费下载链接】mongoThe MongoDB Database项目地址: https://gitcode.com/GitHub_Trending/mo/mongo

导读

本文基于 MongoDB 服务器源码仓库中 README_router_role_api.md 展开，系统讲解 MongoDB 分片集群中Router Role API的设计思想与使用方法：从CollectionRouter/DBPrimaryRouter/MultiCollectionRouter三个核心路由类，到RoutingContext的版本化路由上下文，再到Scatter-Gather命令分发与ShardVersion版本附加 API。读完本文，你将掌握在mongos侧如何正确编排"获取路由信息 → 定位目标分片 → 附加版本元数据 → 分发命令 → 处理 stale 错误并重试"的完整流程，并能够将这套模式复用到新的 router 侧命令开发中。

Router Role 是什么：与 Shard Role 的职责分野

在 MongoDB 的分片集群架构中，任何需要把操作路由到合适分片的代码，都运行在Router Role中。与之相对的是Shard Role——后者直接访问数据集合，运行在各分片（shard）节点上。

Router Role 下的操作必须完成三件事：

1. 获取目标集合（collection）或目标数据库主分片（DBPrimary）的路由信息；
2. 依据路由信息把请求分发到正确的分片；
3. 当某个分片因为路由节点（mongos）上的路由信息过期（stale）而返回错误时，刷新来自 config server 的路由数据，并重试整个请求。

从源码看，这一职责边界非常清晰：router_role.h 顶部注释明确指出，两个路由类"声明了其route方法执行的 scope 是相关数据库或集合的 router"，并且"这些类是目前获取给定条目路由信息的唯一途径"。

CollectionRouter 与 DBPrimaryRouter：两类最基础的路由器

router_role.h 提供了CollectionRouter和DBPrimaryRouter两个类，分别负责：

• CollectionRouter：将命令路由到拥有集合数据的分片。其类注释说明它"主要用于路由 CRUD 操作，这些操作需要看到集合的完整路由表"。
• DBPrimaryRouter：将命令路由到数据库的主分片（DBPrimary shard）。其类注释说明它"主要用于路由需要从数据库主分片协调的 DDL 操作"。

两者共享基类RouterBase（持有OperationContext* _opCtx与CatalogCache* _catalogCache），并都实现了"while (true)重试循环 +_onException异常处理"的核心骨架。

基本用法示例

// CollectionRouter：面向集合，回调收到 RoutingContext sharding::router::CollectionRouter router(opCtx, nss); return router.routeWithRoutingContext( "<Comment to identify this process>"sd, & { ... // 使用 routingCtx 分发一个 collection 请求 ... } );

// DBPrimaryRouter：面向数据库主分片，回调收到 CachedDatabaseInfo sharding::router::DBPrimaryRouter router(opCtx, nss.dbName()); return router.route( "<Comment to identify this process>"sd, & { ... // 使用 dbInfo 分发一个 DBPrimary 请求 ... } );

注意routeWithRoutingContext与route的差异：前者回调内收到的是RoutingContext&（见下文 RoutingContext 章节），后者收到的是CachedDatabaseInfo或CollectionRoutingInfo（面向 DBPrimary 时是数据库缓存信息）。CollectionRouter同时提供route()（回调收CollectionRoutingInfo）与routeWithRoutingContext()（回调收RoutingContext）两种入口，从 router_role.h 源码可见routeWithRoutingContext内部隐式调用了routing_context_utils::runAndValidate，会在一轮操作结束后强制校验路由表。

仓库中的真实调用示例

原文档给出两个真实用例，在仓库中均可定位到对应实现：

• CollectionRouter 用例：原文档引用的rename_collection_coordinator.cpp（在分片集群重命名集合时，需要为config.system.sessions集合在其数据所在的所有分片上创建索引）。该文件在当前仓库中已重构/迁移，但同样的模式在 resharding_recipient_service_external_state.cpp 中有大量同类体现，例如getCollectionIndexes()使用CollectionRouter定位持有全局最小 chunk 的分片加载索引：

sharding::router::CollectionRouter router(opCtx, nss); return router.route(reason, & { uassert(ErrorCodes::NamespaceNotFound, str::stream() << "Expected collection " << nss.toStringForErrorMsg() << " to be tracked", cri.hasRoutingTable()); return MigrationDestinationManager::getCollectionIndexes(opCtx, nss, ...); });

• DBPrimaryRouter 用例：分片集群中删除 resharding 临时集合等操作必须只发往 DBPrimary，因为DBPrimary 负责实例化 ShardingCoordinator，由它跨所有分片协调整个 DDL 操作。这在 resharding_recipient_service_external_state.cpp 中也有对应实现：

// Load the collection options from the primary shard for the database. sharding::router::DBPrimaryRouter router(opCtx, nss.dbName()); return router.route(reason, & { return MigrationDestinationManager::getCollectionOptions( opCtx, NamespaceStringOrUUID{nss.dbName(), uuid}, cdb->getPrimary(), cdb->getVersion(), afterClusterTime); });

关于分片集群 DDL 操作的完整机制，可进一步阅读 README_ddl_operations.md。

路由器内部处理的三个环节

两个类在内部统一完成以下流程（这也是 Router Role 的核心价值所在）：

1. 获取路由信息：为指定集合或 DBPrimary 分片获取路由信息，并以RoutingContext（集合场景）或CachedDatabaseInfo（数据库场景）形式传入 lambda 回调；
2. 检测并处理 stale 路由错误：若分片响应表明路由数据过期，自动刷新路由数据并重试整个操作；
3. 结束后校验：操作成功后对RoutingContext做校验，确保所有声明的命名空间都通过版本化请求向分片验证过。校验的具体不变量见 README_routing_context.md 的 Invariants 一节。

从 router_role.cpp 的实现细节看，stale 错误处理非常精细。CollectionRouterCommon::_onException依据错误码分流：

• StaleDbVersion：调用_catalogCache->onStaleDatabaseVersion()刷新数据库版本；
• StaleConfig：先判断 stale 的命名空间是否属于本次路由涉及的命名空间（含 time-series buckets 与 view 命名空间的相互转换场景），再调用_catalogCache->onStaleCollectionVersion()刷新，并通过staleConfigRetryAttempt计数重试次数；
• StaleEpoch：根据是否有StaleEpochInfo决定刷新单集合还是所有目标集合；
• ShardNotFound：说明分片已被移除，刷新所有目标集合与数据库的路由信息后重试；
• TransactionParticipantFailedUnyield：提取原始错误，若原始错误为 stale 类型则先刷新缓存再抛错。

同时，RouterBase::_initTxnRouterIfNeeded()保证在多文档事务场景下正确接入TransactionRouter，而_armStaleConfigRetryAttemptTracking()会在重试循环开始前把 StaleConfig 重试计数器置 0（嵌套 router 不会重置外层已递增的计数）。所有重试都受gMaxNumStaleVersionRetries服务器参数约束，超过上限会以"Exceeded maximum number of X retries attempting '<comment>'"报错。此外，事务内不允许重试 stale 错误——_onException末尾会检查TransactionRouter::get(_opCtx)，存在事务路由器时直接uassertStatusOK(s)抛出。

使用 Router 时的三条铁律

原文档明确强调，使用CollectionRouter或DBPrimaryRouter时必须遵守：

1. 必须使用回调提供的RoutingContext/CachedDatabaseInfo对象来向分片分发带 shard 版本的命令，推荐配合下文介绍的 Scatter-Gather API；
2. 分片返回的任何 stale 路由错误都必须抛出（throw），由 router 逻辑统一捕获、刷新并重试；自行吞掉错误会破坏版本一致性；
3. 单次路由操作内只能查阅一个版本的路由表——这是保证一致性的关键约束，绝不能在多次访问中混用不同版本的路由信息。

关于版本协议（shard versioning / db versioning）的更深入原理，可阅读 README_versioning_protocols.md 架构指南。

RoutingContext：路由上下文的不可变快照

CollectionRouter::routeWithRoutingContext的回调之所以收到RoutingContext，是因为它承载了"路由操作"这一抽象。RoutingContext在构造时一次性从CatalogCache获取所有声明的命名空间的路由表，并在整个操作期间保持不可变。其完整说明见 README_routing_context.md 与 routing_context.h。

RoutingContext的 consistency invariants（不变量）包括：

• 操作所需的所有路由表都在RoutingContext构造时获取，之后不可变、不可新增；
• 路由操作只能访问构造时预先声明的nss的路由表；
• RoutingContext在以下三种情况之一成立时才能安全终止：
  1. 所有声明的命名空间都已通过向分片发送版本化请求完成路由表验证；
  2. 调用方显式调用skipValidation()（仅当路由表只用于性能优化、不参与查询正确性决策时允许）；
  3. 抛出了 stale 路由元数据异常（如集合 generation 已变化）并向上传播。

未经验证就终止属于逻辑 bug：测试环境下服务器会tassert，生产环境下则打印错误日志。

关键 API 有三个（见 routing_context.h 中的routing_context_utils命名空间）：

routing_context_utils::runAndValidate的源码逻辑很直白：回调正常返回后立即调用routingCtx.validateOnContextEnd()，无论回调返回 void 还是值。而_getCollectionRoutingInfo在 snapshot 读关注且设置了atClusterTime时，会调用CatalogCache::getCollectionRoutingInfoAt获取某个时间点的历史路由表，保证因果一致性。

MultiCollectionRouter：一次路由循环处理多个集合

MultiCollectionRouter在CollectionRouter基础上扩展了能力：在单个 router 重试循环内路由到多个集合。它的典型场景是聚合管道中包含多个$lookup阶段——这些阶段会在同一执行上下文内查询不同的 foreign collection，只要其中任何一个集合路由信息过期，整个操作就必须整体重试。若逐一单独路由，会破坏多集合间的一致性视图。

std::vector<NamespaceString> nssList{nss1, nss2}; sharding::router::MultiCollectionRouter multiCollectionRouter( opCtx->getServiceContext(), nssList ); multiCollectionRouter.route( "<Comment to identify this process>"sd, & { ... // 使用 criMap 分发命令 ... } );

从 router_role.cpp 的实现看，MultiCollectionRouter::route的重试循环每次迭代都会为_targetedNamespaces中的每一个nss 调用_getRoutingInfo(nss)构建criMap，再执行回调；任何一个集合抛出的 stale 错误都会由_onException统一处理。注意_getRoutingInfo会透传allowLocks（事务内允许从已持锁的 CatalogCache 读取）以及atClusterTime时间点路由信息。

仓库中的真实用例位于 initialize_auto_get_helper.h：聚合管道初始化$lookup的自动获取（auto-get）逻辑时，用MultiCollectionRouter一次性获取主集合与所有 secondary 集合的CollectionRoutingInfo，并通过multiCollectionRouter.isAnyCollectionNotLocal(opCtx, criMap)判断是否存在非本地集合，从而决定能否将$lookup下推（pushdown）到 SBE 执行。isAnyCollectionNotLocal的实现会逐集合判断：分片集合必然非本地；不可拆分（unsplittable）集合仅当 MinKey chunk 属于本分片才算本地；未跟踪集合仅当本分片是数据库主分片才算本地。

Scatter-Gather API：版本化命令的分发与聚合

Router Role API 管理的是高层工作流（路由上下文生命周期、重试逻辑、校验），而命令在分片间的实际定位、版本附加与分发由 cluster_commands_helpers.h 中定义的Scatter-Gather API完成。scatter-gather 系列函数提供了"向多个分片并行分发版本化命令并聚合响应"的高层抽象。

scatterGatherVersionedTargetByRoutingTable：按路由表自动定位

该函数依据查询定位逻辑决定命令发往哪些分片；如果查询为空，则命令发往持有该集合 chunk 的所有分片。

std::vector<AsyncRequestsSender::Response> scatterGatherVersionedTargetByRoutingTable( OperationContext* opCtx, RoutingContext& routingCtx, const NamespaceString& nss, const BSONObj& cmdObj, const ReadPreferenceSetting& readPref, Shard::RetryPolicy retryPolicy, const BSONObj& query, const BSONObj& collation // ... 其他参数：letParameters / runtimeConstants / eligibleForSampling / executor );

工作流：

1. 调用buildVersionedRequestsForTargetedShards()：
   • 将查询与路由表（CollectionRoutingInfo+ChunkManager）比对分析；
   • 确定哪些分片拥有匹配数据；
   • 为每个目标分片构建带版本信息的命令对象；
2. 通过gatherResponses()并行分发命令；
3. 返回聚合后的响应。

源码侧的关键支撑有：getVersionedRequestsForTargetedShards()（依据查询与 collation 计算出std::set<ShardId>并为每个 shard 构造请求，见 cluster_commands_helpers.h）以及gatherResponses()（并行分发全部请求并等待完成，若任一分片返回 StaleConfig 则直接抛出该错误，无论其他错误是什么——这正是 router 重试机制得以工作的前提）。此外还有buildVersionedCommandsByRoutingTable()这一模板化的 typed 版本，供使用AsyncRPC的调用方以CommandType形式构建命令。注意：函数声明标注了[[nodiscard]]且"不会在 StaleConfig 错误上重试"——重试职责归 Router Role API 的routeWithRoutingContext循环。

与 Router Role API 组合的完整示例（原文档代码，参数顺序以仓库头文件声明为准）：

#include "src/mongo/db/router_role/router_role.h" #include "src/mongo/db/router_role/cluster_commands_helpers.h" // Contains utility APIs // Complete router operation using all API layers StatusWith<BSONObj> executeShardedQuery( OperationContext* opCtx, const NamespaceString& nss, const BSONObj& query) { // ROUTER ROLE API: Set up routing workflow sharding::router::CollectionRouter router(opCtx, nss); return router.routeWithRoutingContext( "Complete sharded query example", & { // SCATTER-GATHER API: Automated targeting and dispatch auto responses = scatterGatherVersionedTargetByRoutingTable( opCtx, routingCtx, // From Router Role API nss, BSON("find" << nss.coll()), ReadPreferenceSetting(ReadPreference::PrimaryPreferred), Shard::RetryPolicy::kIdempotent, query, BSONObj() ); // Internally, scatter-gather uses: // - QUERY TARGETING API to determine shards // - SHARD VERSIONING API to attach versions // Process results return mergeShardResponses(responses); } ); // Router Role API handles stale routing errors and retries }

scatterGatherVersionedTargetToShards：显式指定目标分片

该函数绕过查询分析，直接对调用方显式指定的分片集合执行版本化命令：

std::vector<AsyncRequestsSender::Response> scatterGatherVersionedTargetToShards( OperationContext* opCtx, RoutingContext& routingCtx, const DatabaseName& dbName, const NamespaceString& nss, const BSONObj& cmdObj, const ReadPreferenceSetting& readPref, Shard::RetryPolicy retryPolicy, const std::set<ShardId>& targetShards );

（仓库 cluster_commands_helpers.h 中的实际签名还包含可选的eligibleForSampling参数。）

适用场景：

• 调用方已经自行确定目标分片集合；
• 需要对分片定位做细粒度控制的操作。

使用示例：

#include "src/mongo/db/router_role/router_role.h" #include "src/mongo/db/router_role/cluster_commands_helpers.h" // Contains utility APIs StatusWith<BSONObj> executeShardedQuery( OperationContext* opCtx, const NamespaceString& nss, const BSONObj& query) { sharding::router::CollectionRouter router(opCtx, nss); return router.routeWithRoutingContext( "Complete targeted sharded query example", & { // Custom targeting logic beyond standard chunk-based routing auto targetedShardsSet = computeShardsToTargetForSpecialCase(routingCtx); // SCATTER-GATHER API: Explicitly target computed shard set auto response = scatterGatherVersionedTargetToShards( opCtx, routingCtx, // From Router Role API DatabaseName::kAdmin, // Custom database name nss, targetedShardsSet, BSON("find" << nss.coll()), ReadPreferenceSetting(ReadPreference::PrimaryPreferred), Shard::RetryPolicy::kIdempotent, false // eligibleForSampling ).front(); return response; } ); // Router Role API handles stale routing errors and retries }

说明：原文档中的该示例存在参数顺序与数量瑕疵（将nss与targetedShardsSet混排在DatabaseName与cmdObj之间），以上代码已按 cluster_commands_helpers.h 中的真实签名（dbName, nss, shards, cmdObj, readPref, retryPolicy）修正，实际开发请以头文件声明为准。

何时才应使用底层 API

绝大多数 router 侧操作都应使用高层 scatter-gather 函数。直接使用buildVersionedRequests/gatherResponses等底层 API仅在特殊情况下被允许：

• 需要自定义分片定位逻辑的复杂聚合管道；
• 需要对请求构建做细粒度控制的操作；
• 标准定位不适用的情况（例如sharded_agg_helpers.cpp使用 RemoteCursor API 的场景）。

Shard Versioning API：统一附加版本元数据

所有面向分片集合的 router 侧操作都必须携带版本元数据，以保障路由一致性并检测过期元数据。请使用标准化的appendShardVersion函数（见 cluster_commands_helpers.h）：

// Append shard version to an existing command object BSONObj appendShardVersion(BSONObj cmdObj, ShardVersion version); // Append shard version to a BSONObjBuilder void appendShardVersion(BSONObjBuilder& cmd, ShardVersion version);

使用示例：

BSONObj cmd = BSON("find" << "myCollection"); auto versionedCmd = appendShardVersion(std::move(cmd), routingCtx.getShardVersion(shardId));

重要准则：

• 绝不手动序列化版本信息；
• 始终使用appendShardVersion函数，以保证字段命名一致与BSON 序列化正确；
• 确保在向分片集合发送任何命令之前附加版本。

从实现侧看，版本附加是分层完成的：router_role.cpp 中的CollectionRouterCommon::appendCRUDRoutingTokenToCommand会在ShardVersion::UNTRACKED()（未跟踪版本，即未分片集合）时额外附加数据库版本（若数据库版本非 Fixed），否则只附加cri.getShardVersion(shardId)；DBPrimaryRouter::appendDDLRoutingTokenToCommand与appendCRUDUnshardedRoutingTokenToCommand则分别面向 DDL 路由令牌与未分片 CRUD 路由令牌。此外还有appendDbVersionIfPresent()系列工具（注意其注释提示：IDL 生成的 typed 命令应优先使用generic_argument_util::setDbVersionIfPresent()），以及applyReadWriteConcern/setReadWriteConcern用于把 OpCtx 上的读写关注应用到发往分片的命令上。

架构分层总结：三层 API 的协作关系

Router Role 由三个互补的 API 层构成，自顶向下逐层调用：

┌─────────────────────────────────────────────┐ │ ROUTER ROLE API │ │ - CollectionRouter / DBPrimaryRouter / │ │ MultiCollectionRouter │ │ - 管理 RoutingContext 生命周期 │ │ - 检测 stale 路由错误并重试 │ │ - 操作结束后校验 RoutingContext │ └──────────────────┬──────────────────────────┘ │ provides RoutingContext to ▼ ┌─────────────────────────────────────────────┐ │ SCATTER-GATHER API（命令分发） │ │ - 分析查询确定目标分片 │ │ - 并发构建并分发版本化命令 │ │ - 聚合多个分片的响应 │ └──────────────────┬──────────────────────────┘ │ uses ▼ ┌─────────────────────────────────────────────┐ │ SHARD VERSIONING API │ │ - 保证 ShardVersion 附加的一致性 │ │ - 防止手动序列化版本信息 │ │ - 提供版本控制的单一控制点 │ └─────────────────────────────────────────────┘

• Router Role API负责"大局"：路由上下文的获取与生命周期、stale 错误检测与重试、操作结束后的校验；
• Scatter-Gather API负责"落地"：查询分析定位分片、构建带版本命令、并行分发与响应聚合；
• Shard Versioning API负责"细节"：确保版本字段以统一方式附加，杜绝手工序列化带来的不一致。

三层协作的完整链路即：CollectionRouter::routeWithRoutingContext构造并校验RoutingContext→ 回调内调用scatterGatherVersionedTargetByRoutingTable（其内部经getShardIdsForQuery定位分片、appendShardVersion附加版本、gatherResponses分发聚合）→ 若分片返回StaleConfig/StaleDbVersion等错误，router 的_onException刷新CatalogCache对应条目并重试整个操作，直至成功或超过gMaxNumStaleVersionRetries上限。

参考与延伸阅读

• Router Role API 主文档：README_router_role_api.md
• RoutingContext 不变量与工具函数：README_routing_context.md、routing_context.h
• 路由器类实现：router_role.h、router_role.cpp
• Scatter-Gather 与版本附加 API：cluster_commands_helpers.h、cluster_commands_helpers.cpp
• 单元测试（含MockRoutingContext用法）：router_role_test.cpp、routing_table_cache_gossip_metadata_hook_test.cpp
• 真实调用方：聚合$lookup自动获取 initialize_auto_get_helper.h、resharding 外部状态 resharding_recipient_service_external_state.cpp、均衡器 moveRange balancer.cpp
• 扩展阅读：分片 DDL 操作 README_ddl_operations.md、版本协议架构 README_versioning_protocols.md

【免费下载链接】mongoThe MongoDB Database项目地址: https://gitcode.com/GitHub_Trending/mo/mongo