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 下的操作必须完成三件事获取目标集合collection或目标数据库主分片DBPrimary的路由信息依据路由信息把请求分发到正确的分片当某个分片因为路由节点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 processsd, { ... // 使用 routingCtx 分发一个 collection 请求 ... } );// DBPrimaryRouter面向数据库主分片回调收到 CachedDatabaseInfo sharding::router::DBPrimaryRouter router(opCtx, nss.dbName()); return router.route( Comment to identify this processsd, { ... // 使用 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 的核心价值所在获取路由信息为指定集合或 DBPrimary 分片获取路由信息并以RoutingContext集合场景或CachedDatabaseInfo数据库场景形式传入 lambda 回调检测并处理 stale 路由错误若分片响应表明路由数据过期自动刷新路由数据并重试整个操作结束后校验操作成功后对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时必须遵守必须使用回调提供的RoutingContext/CachedDatabaseInfo对象来向分片分发带 shard 版本的命令推荐配合下文介绍的 Scatter-Gather API分片返回的任何 stale 路由错误都必须抛出throw由 router 逻辑统一捕获、刷新并重试自行吞掉错误会破坏版本一致性单次路由操作内只能查阅一个版本的路由表——这是保证一致性的关键约束绝不能在多次访问中混用不同版本的路由信息。关于版本协议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在以下三种情况之一成立时才能安全终止所有声明的命名空间都已通过向分片发送版本化请求完成路由表验证调用方显式调用skipValidation()仅当路由表只用于性能优化、不参与查询正确性决策时允许抛出了 stale 路由元数据异常如集合 generation 已变化并向上传播。未经验证就终止属于逻辑 bug测试环境下服务器会tassert生产环境下则打印错误日志。关键 API 有三个见 routing_context.h 中的routing_context_utils命名空间API用途适用场景withValidatedRoutingContext(opCtx, nssList, fn)自行构造RoutingContext→ 执行回调 → 结束后校验非幂等操作的首选withValidatedRoutingContextForTxnCmd变体会额外检查事务中是否允许持锁runAndValidate(routingCtx, fn)对已存在的RoutingContext执行回调并调用validateOnContextEnd()例如从CollectionRoutingInfoTargeter复用已构造的上下文CollectionRouter::routeWithRoutingContext(comment, fn)构造 → 执行 → 遇 stale 错误刷新缓存并换新RoutingContext重试读与幂等操作的首选内部隐式调用withValidatedRoutingContextrouting_context_utils::runAndValidate的源码逻辑很直白回调正常返回后立即调用routingCtx.validateOnContextEnd()无论回调返回 void 还是值。而_getCollectionRoutingInfo在 snapshot 读关注且设置了atClusterTime时会调用CatalogCache::getCollectionRoutingInfoAt获取某个时间点的历史路由表保证因果一致性。MultiCollectionRouter一次路由循环处理多个集合MultiCollectionRouter在CollectionRouter基础上扩展了能力在单个 router 重试循环内路由到多个集合。它的典型场景是聚合管道中包含多个$lookup阶段——这些阶段会在同一执行上下文内查询不同的 foreign collection只要其中任何一个集合路由信息过期整个操作就必须整体重试。若逐一单独路由会破坏多集合间的一致性视图。std::vectorNamespaceString nssList{nss1, nss2}; sharding::router::MultiCollectionRouter multiCollectionRouter( opCtx-getServiceContext(), nssList ); multiCollectionRouter.route( Comment to identify this processsd, { ... // 使用 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::vectorAsyncRequestsSender::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 );工作流调用buildVersionedRequestsForTargetedShards()将查询与路由表CollectionRoutingInfoChunkManager比对分析确定哪些分片拥有匹配数据为每个目标分片构建带版本信息的命令对象通过gatherResponses()并行分发命令返回聚合后的响应。源码侧的关键支撑有getVersionedRequestsForTargetedShards()依据查询与 collation 计算出std::setShardId并为每个 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 StatusWithBSONObj 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::vectorAsyncRequestsSender::Response scatterGatherVersionedTargetToShards( OperationContext* opCtx, RoutingContext routingCtx, const DatabaseName dbName, const NamespaceString nss, const BSONObj cmdObj, const ReadPreferenceSetting readPref, Shard::RetryPolicy retryPolicy, const std::setShardId 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 StatusWithBSONObj 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.mdRoutingContext 不变量与工具函数README_routing_context.md、routing_context.h路由器类实现router_role.h、router_role.cppScatter-Gather 与版本附加 APIcluster_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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 MongoDB Router Role API 深度解析：CollectionRouter 路由框架与 Scatter-Gather 命令分发机制
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 下的操作必须完成三件事获取目标集合collection或目标数据库主分片DBPrimary的路由信息依据路由信息把请求分发到正确的分片当某个分片因为路由节点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 processsd, { ... // 使用 routingCtx 分发一个 collection 请求 ... } );// DBPrimaryRouter面向数据库主分片回调收到 CachedDatabaseInfo sharding::router::DBPrimaryRouter router(opCtx, nss.dbName()); return router.route( Comment to identify this processsd, { ... // 使用 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 的核心价值所在获取路由信息为指定集合或 DBPrimary 分片获取路由信息并以RoutingContext集合场景或CachedDatabaseInfo数据库场景形式传入 lambda 回调检测并处理 stale 路由错误若分片响应表明路由数据过期自动刷新路由数据并重试整个操作结束后校验操作成功后对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时必须遵守必须使用回调提供的RoutingContext/CachedDatabaseInfo对象来向分片分发带 shard 版本的命令推荐配合下文介绍的 Scatter-Gather API分片返回的任何 stale 路由错误都必须抛出throw由 router 逻辑统一捕获、刷新并重试自行吞掉错误会破坏版本一致性单次路由操作内只能查阅一个版本的路由表——这是保证一致性的关键约束绝不能在多次访问中混用不同版本的路由信息。关于版本协议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在以下三种情况之一成立时才能安全终止所有声明的命名空间都已通过向分片发送版本化请求完成路由表验证调用方显式调用skipValidation()仅当路由表只用于性能优化、不参与查询正确性决策时允许抛出了 stale 路由元数据异常如集合 generation 已变化并向上传播。未经验证就终止属于逻辑 bug测试环境下服务器会tassert生产环境下则打印错误日志。关键 API 有三个见 routing_context.h 中的routing_context_utils命名空间API用途适用场景withValidatedRoutingContext(opCtx, nssList, fn)自行构造RoutingContext→ 执行回调 → 结束后校验非幂等操作的首选withValidatedRoutingContextForTxnCmd变体会额外检查事务中是否允许持锁runAndValidate(routingCtx, fn)对已存在的RoutingContext执行回调并调用validateOnContextEnd()例如从CollectionRoutingInfoTargeter复用已构造的上下文CollectionRouter::routeWithRoutingContext(comment, fn)构造 → 执行 → 遇 stale 错误刷新缓存并换新RoutingContext重试读与幂等操作的首选内部隐式调用withValidatedRoutingContextrouting_context_utils::runAndValidate的源码逻辑很直白回调正常返回后立即调用routingCtx.validateOnContextEnd()无论回调返回 void 还是值。而_getCollectionRoutingInfo在 snapshot 读关注且设置了atClusterTime时会调用CatalogCache::getCollectionRoutingInfoAt获取某个时间点的历史路由表保证因果一致性。MultiCollectionRouter一次路由循环处理多个集合MultiCollectionRouter在CollectionRouter基础上扩展了能力在单个 router 重试循环内路由到多个集合。它的典型场景是聚合管道中包含多个$lookup阶段——这些阶段会在同一执行上下文内查询不同的 foreign collection只要其中任何一个集合路由信息过期整个操作就必须整体重试。若逐一单独路由会破坏多集合间的一致性视图。std::vectorNamespaceString nssList{nss1, nss2}; sharding::router::MultiCollectionRouter multiCollectionRouter( opCtx-getServiceContext(), nssList ); multiCollectionRouter.route( Comment to identify this processsd, { ... // 使用 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::vectorAsyncRequestsSender::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 );工作流调用buildVersionedRequestsForTargetedShards()将查询与路由表CollectionRoutingInfoChunkManager比对分析确定哪些分片拥有匹配数据为每个目标分片构建带版本信息的命令对象通过gatherResponses()并行分发命令返回聚合后的响应。源码侧的关键支撑有getVersionedRequestsForTargetedShards()依据查询与 collation 计算出std::setShardId并为每个 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 StatusWithBSONObj 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::vectorAsyncRequestsSender::Response scatterGatherVersionedTargetToShards( OperationContext* opCtx, RoutingContext routingCtx, const DatabaseName dbName, const NamespaceString nss, const BSONObj cmdObj, const ReadPreferenceSetting readPref, Shard::RetryPolicy retryPolicy, const std::setShardId 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 StatusWithBSONObj 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.mdRoutingContext 不变量与工具函数README_routing_context.md、routing_context.h路由器类实现router_role.h、router_role.cppScatter-Gather 与版本附加 APIcluster_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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考