消息队列后端微服务流处理【免费下载链接】rocketmqApache RocketMQ is a cloud native messaging and streaming platform, making it simple to build event-driven applications.项目地址https://gitcode.com/gh_mirrors/ro/rocketmq点击查看免费下载导读在 Apache RocketMQ 的 Controller自动主备切换模式下Broker 节点需要在整个集群生命周期内拥有稳定、唯一的身份标识才能保证ReplicaInfo、SyncStateSet等主备元数据在节点重启、IP 漂移后依然可关联。本文以 docs/cn/controller/persistent_unique_broker_id.md 为主线系统讲解 5.1.1 版本引入的持久化唯一 BrokerId机制从设计动机、协商上线流程、故障容错到从 5.0.0/5.1.0 的升级步骤与版本兼容性并结合仓库源码ReplicasManager.java、ReplicasInfoManager.java佐证底层实现。读完本文你将掌握该机制的工作原理、关键文件与参数以及如何在容器/K8s 环境下正确升级与运维。一、背景为什么需要持久化唯一 BrokerId在 RocketMQ 5.0.0 与 5.1.0 版本中Controller 模式下使用BrokerAddressIP:Port作为 Broker 在 Controller 侧的唯一标识。这一设计在容器化K8s环境中存在明显缺陷容器或 Pod 每次重启、升级都可能导致 IP 发生变化IP 变化后Controller 上以旧BrokerAddress记录的数据如ReplicaInfo、SyncStateSet等无法与重启后的新 Broker 对应起来主备关系与同步状态难以恢复。因此从 5.1.1 开始Controller 侧改用BrokerName:BrokerId作为 Broker 的唯一标识不再以BrokerAddress为唯一标识。由于ClusterName与BrokerName都是启动时在配置文件中固定配置的真正的核心问题只剩两个BrokerId的分配何时给谁分配哪个 IDBrokerId的持久化保证重启后不丢失、不重复。1.1 BrokerId 的分配规则BrokerId从1开始分配对应源码常量MixAll.FIRST_BROKER_CONTROLLER_ID 1L见 MixAll.java在同一BrokerName即同一个 broker-set内递增且唯一当一个 Broker 被选举为 Master 时向 Name Server 注册时将使用BrokerId 0MixAll.MASTER_ID 0L以兼容既有逻辑——brokerId 为 0 代表 Master 身份也就是说Controller 侧用正整数 BrokerId 唯一标识节点Name Server 侧仍用 0 标识 Master二者通过被选为 Master 的节点建立对应关系。默认持久化 BrokerId 的文件位于~/store/brokerIdentity可通过storePathBrokerIdentity参数自定义存储路径。在自动主备切换模式下不要随意删除该文件否则该 Broker 会被当作全新 Broker 上线。二、首次上线流程Broker 与 Controller 如何协商出唯一 BrokerIdBroker 第一次上线时本地只有配置文件中的ClusterName、BrokerName以及自身的BrokerAddress还没有BrokerId。整个协商流程分为 6 步Broker 侧的完整实现位于 ReplicasManager.register()Controller 侧的状态机实现位于 ReplicasInfoManager.java。第 1 步GetNextBrokerId Request / ReadFromDLedgerBroker 向 Controller 发起GetNextBrokerId请求获取当前该 broker-set 中下一个待分配的 BrokerId从 1 开始。请求头为 GetNextBrokerIdRequestHeader携带clusterName与brokerNameController 收到后经由 DLedgerRaft 协议读取状态机中的NextBrokerId数据对应源码 ReplicasInfoManager.getNextBrokerId()若该 broker-set 尚未注册任何 Broker则返回FIRST_BROKER_CONTROLLER_ID即 1否则返回getNextAssignBrokerId()。第 2 步GetNextBrokerId Response / CreateTempMetaFileController 将NextBrokerId返回给 Broker。Broker 拿到后生成一个RegisterCode注册校验码格式为$ipAddress;$timestamp见 ReplicasManager.createTempMetadataFile()用于后续的身份校验创建临时文件.broker.meta.temp实际路径为brokerIdentity-temp见 ReplicasManager 构造函数其中持久化记录期望应用的BrokerId与RegisterCode。临时元数据的内容格式由 TempBrokerMetadata 定义以#分隔clusterName#brokerName#brokerId#registerCheckCode第 3 步ApplyBrokerId Request / CASApplyBrokerIdBroker 携带自身基本数据ClusterName、BrokerName、BrokerAddress以及期望应用的BrokerId、RegisterCode向 Controller 发送ApplyBrokerId请求。Controller 通过 DLedger写入该事件ApplyBrokerIdEvent当事件日志被应用到状态机时执行条件判断相当于 CAS 语义若该BrokerId尚未被分配则分配成功若该BrokerId已被分配且恰好属于该 BrokerregisterCheckCode匹配同样视为成功若该BrokerId已被分配给其他 Broker则失败。对应校验逻辑在 ReplicasInfoManager.applyBrokerId()成功后会记录BrokerId与RegisterCode的绑定关系状态机应用事件的实现是 handleApplyBrokerId()——首次注册时还会为这个 broker-set 初始化空的SyncStateInfo。第 4 步ApplyBrokerId Response / CreateMetaFileFromTemp若上一步成功应用Controller 返回成功此时 Broker 侧可视为已成功分配该 BrokerId若失败Controller 返回当前的NextBrokerIdBroker 需要回退重试见下文故障容错。成功时Broker 需要将 BrokerId 信息彻底持久化原子地删除.broker.meta.temp并创建.broker.metabrokerIdentity文件。这两步必须是原子操作防止出现文件不一致。实现见 ReplicasManager.createMetadataFileAndDeleteTemp()正式元数据由 BrokerMetadata 定义格式为clusterName#brokerName#brokerId此后Broker 将brokerControllerId设置为该值并同步给 HA 服务haService.setLocalBrokerId主从复制AutoSwitchHA开始以该 ID 标识本节点。第 5 步RegisterBrokerToController Request / UpdateBrokerAddress前面步骤已协商出双方认同的BrokerId但 Controller 状态机中保存的BrokerAddress可能是上一次上线时的旧地址需要更新。Broker 发送RegisterBrokerToController请求携带当前的BrokerAddress。Controller 比对当前该 Broker 在状态机中保存的BrokerAddress若与请求中的不一致则通过UpdateBrokerAddressEvent更新见 ReplicasInfoManager.registerBroker() 与 handleUpdateBrokerAddress()。第 6 步RegisterBrokerToController ResponseController 更新完BrokerAddress后响应中携带当前该 Broker 所在 broker-set 的主从信息masterBrokerId、masterAddress、masterEpoch、syncStateSetEpoch与SyncStateSet用于通知 Broker 进行相应的身份转变。Broker 侧处理见 ReplicasManager.registerBrokerToController()若自己是 master 则调用changeToMaster否则调用changeToSlave。经过上述流程第一次上线的 Broker 与 Controller 成功协商出一个双方都认同的 BrokerId 并持久化保存IP 变化不再影响身份识别。2.1 Controller 侧协议入口上述请求在 Controller 端由 ControllerRequestProcessor 分发处理CONTROLLER_GET_NEXT_BROKER_ID→handleGetNextBrokerIdCONTROLLER_APPLY_BROKER_ID→handleApplyBrokerIdCONTROLLER_REGISTER_BROKER→handleRegisterBroker。写事件ApplyBrokerId、RegisterBroker会经 DLedger 落盘并由状态机顺序应用保证多副本 Controller 数据一致相关事件定义见 ApplyBrokerIdEvent 与 EventType。三、注册状态轮转与失败恢复Broker 侧的注册流程由RegisterState状态机驱动ReplicasManager.javaINITIAL → CREATE_TEMP_METADATA_FILE_DONE → CREATE_METADATA_FILE_DONE → REGISTERED每次register()调用前会通过confirmNowRegisteringState()读取本地broker.meta/.broker.meta.temp文件来判断当前所处阶段源码从而决定从哪一步继续这正是断点续传式容错的基础。register()的完整判定逻辑源码可概括为确认当前注册状态校验本地 metadata/tempMetadata 与配置中的clusterName、brokerName是否一致checkMetadataValidINITIAL态请求GetNextBrokerId并创建临时元数据文件CREATE_TEMP_METADATA_FILE_DONE态发送ApplyBrokerId成功则创建正式元数据文件并删除临时文件CREATE_METADATA_FILE_DONE态发送RegisterBrokerToController完成注册。如果注册失败startBasicService()会最多重试 5 次每次随机休眠不超过 1 秒以避免并发注册冲突Controller 不可达时还会按RETRY_INTERVAL_SECOND5 秒持续重试源码。四、故障容错各阶段宕机如何保证正确分配如果在正常上线流程中任意阶段发生宕机以下机制保证BrokerId分配的正确性。4.1 正常重启后的节点上线正常重启时双方已协商出唯一BrokerId且本地broker.meta中已有该 ID 的数据因此无需再走注册协商流程直接从RegisterBrokerToController步骤继续上线即可confirmNowRegisteringState会直接判定为CREATE_METADATA_FILE_DONE。4.2 CreateTempMetaFile 失败若上图流程失败Broker 重启后Controller 侧状态机没有分配任何 BrokerIdBroker 自身也没有保存任何数据。此时直接按照完整流程从头开始执行即可不会产生任何残留状态。4.3 CreateTempMetaFile 成功、ApplyBrokerId 未成功Controller 判定本次ApplyBrokerId请求不合法例如请求分配一个已被分配的BrokerId且RegisterCode不相等此时会返回当前的NextBrokerId给 Broker。Broker 的处理是直接删除.broker.meta.temp文件tempBrokerMetadata.clear()将registerState回退为INITIAL回到第 2 步重新执行 GetNextBrokerId 及后续流程。对应实现ReplicasManager.applyBrokerId() 失败分支。4.4 ApplyBrokerId 成功、CreateMetaFileFromTemp 未成功这种情况可能出现在两类故障中ApplyResultDLedger 写日志后的响应丢失——Broker 未收到成功响应但 Controller 状态机实际已应用CAS 删除临时文件并创建broker.meta失败。此时重启后Controller 侧已经认为 ApplyBrokerId 成功状态机中也已修改 BrokerId 分配数据。因此 Broker 不需要重头再来而是直接从步骤 3发送 ApplyBrokerId继续由于本地仍有.broker.meta.temp文件可以从中取出之前成功应用的BrokerId和RegisterCode重新发送给 Controller 后只要 Controller 中存在该BrokerId且RegisterCode与请求中的相等即视为成功随后继续创建正式元数据文件并完成注册。五、正确上线后以 BrokerId 作为唯一标识Broker 正确上线之后后续所有请求与状态记录都以BrokerId作为唯一标识心跳Broker 周期性向 Controller 发送心跳sendHeartbeatToController携带brokerControllerIdSyncStateSet主从同步集合以SetLongBrokerId 集合表示主节点定期检查并上报收缩后的集合doReportSyncStateSetChanged选举与主从切换ElectMaster、GetReplicaInfo等请求均以brokerNamebrokerId定位节点同时 Controller 侧会记录每个BrokerId当前对应的BrokerAddress在主从切换等场景下用于通知 Broker 状态变化例如changeToMaster/changeToSlave中根据masterBrokerId判断身份。5.1 关键配置文件与参数项默认位置 / 值说明正式 BrokerId 文件~/store/brokerIdentity持久化clusterName#brokerName#brokerId由storePathBrokerIdentity决定路径临时 BrokerId 文件brokerIdentity-temp即文档中的.broker.meta.temp持久化clusterName#brokerName#brokerId#registerCheckCode用于崩溃恢复Epoch 文件~/store/epochFileCheckpoint、~/store/epochFileCheckpoint.bak记录主从切换的 Epoch 检查点升级时需要删除storePathBrokerIdentitynull默认拼接storePathRootDir /brokerIdentity自定义 BrokerId 存储路径见 MessageStoreConfig.java校验提示checkMetadataValid()会核对本地元数据中的clusterName、brokerName与配置文件是否一致源码。因此不要将某个 Broker 的brokerIdentity文件复制给另一个 Broker 使用否则会因元数据与配置不匹配而注册失败。六、升级方案从 5.0.0/5.1.0 升级到持久化 BrokerId 版本4.x 版本升级遵守 5.0 升级文档流程即可。5.0.0 和 5.1.0 非持久化 BrokerId 版本升级到 5.1.1 及以上持久化 BrokerId 版本需要按如下流程操作前提当前集群使用 Controller 模式即enableControllerModetrue。6.1 升级 Controller将旧版本 Controller 组整体停机清除 Controller 数据即默认位于~/DLedgerController下的数据文件DLedger 日志其中保存了旧版状态机数据上线新版 Controller 组。在上述升级 Controller 的流程中Broker 仍可正常运行但无法进行主备切换因为 Controller 不可用。6.2 升级 Broker将 Broker从节点停机将 Broker主节点停机删除所有 Broker 的 Epoch 文件即默认的~/store/epochFileCheckpoint和~/store/epochFileCheckpoint.bak先将原先的主 Broker上线等待该 Broker 当选为 master可使用 admin 命令getSyncStateSet观察对应工具类 GetSyncStateSetSubCommand再将原来的从 Broker 全部上线。建议停机时先停从、再停主上线时先上原先的主、再上原先的从这样可以保证原来的主备关系不变。 若需要改变升级前后的主备关系则停机时必须保证主、备的 CommitLog 对齐否则可能导致数据被截断而丢失。6.3 版本兼容性矩阵5.1.0 及以下版本 Controller5.1.1 及以上版本 Controller5.1.0 及以下版本 Broker正常运行可切换若已主备确定则可正常运行不可切换若 Broker 重新启动则无法上线5.1.1 及以上版本 Broker无法正常上线正常运行可切换结论很明确新旧版本必须整体升级不能长期混用。从 5.1.1 开始Broker 与 Controller 必须同时具备持久化 BrokerId 能力才能完成注册与切换升级时应参照上文顺序先升级 Controller清数据、再按主备顺序升级 Broker删 Epoch 文件以最小化停机与数据风险。七、相关文档与源码索引设计文档中/英docs/cn/controller/persistent_unique_broker_id.md、docs/en/controller/persistent_unique_broker_id.mdController 部署与快速开始docs/cn/controller/deploy.md、docs/cn/controller/quick_start.mdBroker 侧注册实现ReplicasManager.java含register()、getNextBrokerId()、applyBrokerId()等Controller 侧状态机ReplicasInfoManager.javaController 基于 DLedger 的实现DLedgerController.java协议请求头remoting/.../controller/register/元数据文件类BrokerMetadata.java、TempBrokerMetadata.java常量定义MixAll.javaMASTER_ID、FIRST_BROKER_CONTROLLER_ID测试参考ReplicasManagerRegisterTest.java、ReplicasInfoManagerTest.java相关集成测试AutoSwitchRoleBase.java赞分享消息队列后端微服务流处理【免费下载链接】rocketmqApache RocketMQ is a cloud native messaging and streaming platform, making it simple to build event-driven applications.项目地址https://gitcode.com/gh_mirrors/ro/rocketmq点击查看免费下载相关推荐RocketMQ Controller 模式下持久化唯一 BrokerId 的设计、注册流程与升级实践RocketMQ Controller 模式下持久化唯一 BrokerId 的设计、注册流程与升级实践 本指南以 docs/cn/controller/pers消息队列流处理后端Apache RocketMQ Controller 模式持久化唯一 BrokerId 设计详解从 BrokerAddress 到 BrokerName:BrokerId 的演进Apache RocketMQ Controller 模式持久化唯一 BrokerId 设计详解从 BrokerAddress 到 BrokerName:Br消息队列流处理后端Apache RocketMQ Controller 模式部署与升级指南从 Controller 部署、Broker 配置到持久化 BrokerID 升级Apache RocketMQ Controller 模式部署与升级指南从 Controller 部署、Broker 配置到持久化 BrokerID 升级 A消息队列后端微服务流处理上一篇MATLAB科研绘图终极指南如何用export_fig导出完美图片的5个秘诀下一篇GitBucket数据库分片方案超大规模团队的存储扩展策略创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考