【RocketMQ】RocketMQ自动容灾切换DLedger集群实践

2023-12-17 20:32
文章标签 集群 实践 切换 自动 rocketmq 容灾 dledger

本文主要是介绍【RocketMQ】RocketMQ自动容灾切换DLedger集群实践，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

文章目录

  • DLedger配置介绍
  • 新建DLedger集群
    • 机器规划
      • Broker
      • nameserver
      • console
    • Broker配置
      • broker-a
      • broker-a-s1
      • broker-a-s2
      • broker-b
      • broker-b-s1
      • broker-b-s2
      • broker-c
      • broker-c-s1
      • broker-c-s2
    • 验证
  • 旧集群升级
    • 杀掉旧的 Broker
    • 检查旧的 Commitlog
    • 修改配置
    • 重新启动 Broker
  • Tips

本文主要演示如何搭建一个多主多从的自动容灾切换的RocketMQ集群，以及如何把现有普通的Master/Slave集群升级为自动容灾切换的集群。

DLedger配置介绍

​ 部署自动容灾的RocketMQ集群和普通的Master/Slave集群没太多差别，只需要在Broker.conf开启DLedger相关配置即可，官方称之为自动容灾切换的RocketMQ-on-DLedger Group，DLedger是分布式一致性协议Raft的一个实现，是一个轻量级的Java类库，RocketMQ引入其来实现Leader选举等一致性操作。

  • RocketMQ-on-DLedger Group是指一组相同名称的 Broker，至少需要 3 个节点，通过 Raft 自动选举出一个 Leader，其余节点 作为 Follower，并在Leader和Follower之间复制数据以保证高可用。
  • RocketMQ-on-DLedger Group能自动容灾切换，并保证数据一致。
  • RocketMQ-on-DLedger Group是可以水平扩展的，也即可以部署任意多个RocketMQ-on-DLedger Group同时对外提供服务。

相关配置

name含义举例
enableDLegerCommitLog是否启动 DLedger，即是否启用RocketMQ主从切换，默认为falsetrue
dLegerGroupDLedger Raft Group的名字，建议和brokerName保持一致RaftNode00
dLegerPeersDLedger Group 内各节点的端口信息，同一个Group 内的各个节点配置必须要保证一致n0-127.0.0.1:40911;n1-127.0.0.1:40912;n2-127.0.0.1:40913
dLegerSelfId节点 id, 必须属于 dLegerPeers 中的一个；同Group 内各个节点要唯一n0
sendMessageThreadPoolNums发送线程个数，建议配置成 Cpu 核数16

新建DLedger集群

机器规划

Broker

节点ID机器端口dLegerPeer
broker-a0172.24.29.21320911n0-172.24.29.213:10911
broker-a-s11172.24.29.21420911n1-172.24.29.214:10911
broker-a-s22172.24.29.21520911n2-172.24.29.215:10911
broker-b0172.24.29.21420915n0-172.24.29.214:10915
broker-b-s11172.24.29.21320915n1-172.24.29.213:10915
broker-b-s22172.24.29.21520915n2-172.24.29.215:10915
broker-c0172.24.29.21520919n0-172.24.29.215:10919
broker-c-s11172.24.29.21320919n1-172.24.29.213:10919
broker-c-s22172.24.29.21420919n2-172.24.29.214:10919

nameserver

节点机器端口
nameserver172.24.29.2159976

console

节点机器端口
console172.24.29.2158585

Broker配置

broker-a

listenPort = 20911
rocketmqHome = /neworiental/rocketmq-test/rocketmq-a/rocketmq471
brokerClusterName = rocketmq-test
namesrvAddr= 172.24.29.215:9976
brokerName = broker-a
#对外提供服务地址
brokerIP1 = 172.24.29.213
#Broker HAIP地址，供slave同步消息的地址
brokerIP2= 172.24.29.213
brokerId = 0
storePathRootDir = /neworiental/rocketmq-test/rocketmq-a/store
storePathCommitLog = /neworiental/rocketmq-test/rocketmq-a/store/commitlog
storePathConsumerQueue = /neworiental/rocketmq-test/rocketmq-a/store/consumequeue
deleteWhen = 04
fileReservedTime = 168
brokerRole = ASYNC_MASTER
flushDiskType = ASYNC_FLUSH
autoCreateTopicEnable = false
#是否自动创建消费组
autoCreateSubscriptionGroup = false
#集群名称是否可用作Topic使用
clusterTopicEnable = false
#Broker名称是否可用作Topic使用
brokerTopicEnable = false
useEpollNativeSelector = true#是否开启ACL
aclEnable=true#多副本自动主从选举Dledger相关
enableDLegerCommitLog = true
dLegerGroup = broker-a
dLegerSelfId = n0
dLegerPeers = n0-172.24.29.213:10911;n1-172.24.29.214:10911;n2-172.24.29.215:10911

broker-a-s1

listenPort = 20911
rocketmqHome = /neworiental/rocketmq-test/rocketmq-a-s1/rocketmq471
brokerClusterName = rocketmq-test
namesrvAddr= 172.24.29.215:9976
brokerName = broker-a
#对外提供服务地址
brokerIP1 = 172.24.29.214
#Broker HAIP地址，供slave同步消息的地址
brokerIP2= 172.24.29.214
brokerId = 1
storePathRootDir = /neworiental/rocketmq-test/rocketmq-a-s1/store
storePathCommitLog = /neworiental/rocketmq-test/rocketmq-a-s1/store/commitlog
storePathConsumerQueue = /neworiental/rocketmq-test/rocketmq-a-s1/store/consumequeue
deleteWhen = 04
fileReservedTime = 168
brokerRole = SLAVE
flushDiskType = ASYNC_FLUSH
autoCreateTopicEnable = false
#是否自动创建消费组
autoCreateSubscriptionGroup = false
#集群名称是否可用作Topic使用
clusterTopicEnable = false
#Broker名称是否可用作Topic使用
brokerTopicEnable = false
useEpollNativeSelector = true#是否开启ACL
aclEnable=true#多副本自动主从选举Dledger相关
enableDLegerCommitLog = true
dLegerGroup = broker-a
dLegerSelfId = n1
dLegerPeers = n0-172.24.29.213:10911;n1-172.24.29.214:10911;n2-172.24.29.215:10911

broker-a-s2

listenPort = 20911
rocketmqHome = /neworiental/rocketmq-test/rocketmq-a-s2/rocketmq471
brokerClusterName = rocketmq-test
namesrvAddr= 172.24.29.215:9976
brokerName = broker-a
#对外提供服务地址
brokerIP1 = 172.24.29.215
#Broker HAIP地址，供slave同步消息的地址
brokerIP2= 172.24.29.215
brokerId = 2
storePathRootDir = /neworiental/rocketmq-test/rocketmq-a-s2/store
storePathCommitLog = /neworiental/rocketmq-test/rocketmq-a-s2/store/commitlog
storePathConsumerQueue = /neworiental/rocketmq-test/rocketmq-a-s2/store/consumequeue
deleteWhen = 04
fileReservedTime = 168
brokerRole = SLAVE
flushDiskType = ASYNC_FLUSH
autoCreateTopicEnable = false
#是否自动创建消费组
autoCreateSubscriptionGroup = false
#集群名称是否可用作Topic使用
clusterTopicEnable = false
#Broker名称是否可用作Topic使用
brokerTopicEnable = false
useEpollNativeSelector = true#是否开启ACL
aclEnable=true#多副本自动主从选举Dledger相关
enableDLegerCommitLog = true
dLegerGroup = broker-a
dLegerSelfId = n2
dLegerPeers = n0-172.24.29.213:10911;n1-172.24.29.214:10911;n2-172.24.29.215:10911

broker-b

listenPort = 20915
rocketmqHome = /neworiental/rocketmq-test/rocketmq-b/rocketmq471
brokerClusterName = rocketmq-test
namesrvAddr= 172.24.29.215:9976
brokerName = broker-b
#对外提供服务地址
brokerIP1 = 172.24.29.214
#Broker HAIP地址，供slave同步消息的地址
brokerIP2= 172.24.29.214
brokerId = 0
storePathRootDir = /neworiental/rocketmq-test/rocketmq-b/store
storePathCommitLog = /neworiental/rocketmq-test/rocketmq-b/store/commitlog
storePathConsumerQueue = /neworiental/rocketmq-test/rocketmq-b/store/consumequeue
deleteWhen = 04
fileReservedTime = 168
brokerRole = ASYNC_MASTER
flushDiskType = ASYNC_FLUSH
autoCreateTopicEnable = false
#是否自动创建消费组
autoCreateSubscriptionGroup = false
#集群名称是否可用作Topic使用
clusterTopicEnable = false
#Broker名称是否可用作Topic使用
brokerTopicEnable = false
useEpollNativeSelector = true#是否开启ACL
aclEnable=true#多副本自动主从选举Dledger相关
enableDLegerCommitLog = true
dLegerGroup = broker-b
dLegerSelfId = n0
dLegerPeers = n0-172.24.29.214:10915;n1-172.24.29.213:10915;n2-172.24.29.215:10915

broker-b-s1

listenPort = 20915
rocketmqHome = /neworiental/rocketmq-test/rocketmq-b-s1/rocketmq471
brokerClusterName = rocketmq-test
namesrvAddr= 172.24.29.215:9976
brokerName = broker-b
#对外提供服务地址
brokerIP1 = 172.24.29.213
#Broker HAIP地址，供slave同步消息的地址
brokerIP2= 172.24.29.213
brokerId = 1
storePathRootDir = /neworiental/rocketmq-test/rocketmq-b-s1/store
storePathCommitLog = /neworiental/rocketmq-test/rocketmq-b-s1/store/commitlog
storePathConsumerQueue = /neworiental/rocketmq-test/rocketmq-b-s1/store/consumequeue
deleteWhen = 04
fileReservedTime = 168
brokerRole = SLAVE
flushDiskType = ASYNC_FLUSH
autoCreateTopicEnable = false
#是否自动创建消费组
autoCreateSubscriptionGroup = false
#集群名称是否可用作Topic使用
clusterTopicEnable = false
#Broker名称是否可用作Topic使用
brokerTopicEnable = false
useEpollNativeSelector = true#是否开启ACL
aclEnable=true#多副本自动主从选举Dledger相关
enableDLegerCommitLog = true
dLegerGroup = broker-b
dLegerSelfId = n1
dLegerPeers = n0-172.24.29.214:10915;n1-172.24.29.213:10915;n2-172.24.29.215:10915

broker-b-s2

listenPort = 20915
rocketmqHome = /neworiental/rocketmq-test/rocketmq-b-s2/rocketmq471
brokerClusterName = rocketmq-test
namesrvAddr= 172.24.29.215:9976
brokerName = broker-b
#对外提供服务地址
brokerIP1 = 172.24.29.215
#Broker HAIP地址，供slave同步消息的地址
brokerIP2= 172.24.29.215
brokerId = 2
storePathRootDir = /neworiental/rocketmq-test/rocketmq-b-s2/store
storePathCommitLog = /neworiental/rocketmq-test/rocketmq-b-s2/store/commitlog
storePathConsumerQueue = /neworiental/rocketmq-test/rocketmq-b-s2/store/consumequeue
deleteWhen = 04
fileReservedTime = 168
brokerRole = SLAVE
flushDiskType = ASYNC_FLUSH
autoCreateTopicEnable = false
#是否自动创建消费组
autoCreateSubscriptionGroup = false
#集群名称是否可用作Topic使用
clusterTopicEnable = false
#Broker名称是否可用作Topic使用
brokerTopicEnable = false
useEpollNativeSelector = true#是否开启ACL
aclEnable=true#多副本自动主从选举Dledger相关
enableDLegerCommitLog = true
dLegerGroup = broker-b
dLegerSelfId = n2
dLegerPeers = n0-172.24.29.214:10915;n1-172.24.29.213:10915;n2-172.24.29.215:10915

broker-c

listenPort = 20919
rocketmqHome = /neworiental/rocketmq-test/rocketmq-c/rocketmq471
brokerClusterName = rocketmq-test
namesrvAddr= 172.24.29.215:9976
brokerName = broker-c
#对外提供服务地址
brokerIP1 = 172.24.29.215
#Broker HAIP地址，供slave同步消息的地址
brokerIP2= 172.24.29.215
brokerId = 0
storePathRootDir = /neworiental/rocketmq-test/rocketmq-c/store
storePathCommitLog = /neworiental/rocketmq-test/rocketmq-c/store/commitlog
storePathConsumerQueue = /neworiental/rocketmq-test/rocketmq-c/store/consumequeue
deleteWhen = 04
fileReservedTime = 168
brokerRole = ASYNC_MASTER
flushDiskType = ASYNC_FLUSH
autoCreateTopicEnable = false
#是否自动创建消费组
autoCreateSubscriptionGroup = false
#集群名称是否可用作Topic使用
clusterTopicEnable = false
#Broker名称是否可用作Topic使用
brokerTopicEnable = false
useEpollNativeSelector = true#是否开启ACL
aclEnable=true#多副本自动主从选举Dledger相关
enableDLegerCommitLog = true
dLegerGroup = broker-c
dLegerSelfId = n0
dLegerPeers = n0-172.24.29.215:10919;n1-172.24.29.213:10919;n2-172.24.29.214:10919

broker-c-s1

listenPort = 20919
rocketmqHome = /neworiental/rocketmq-test/rocketmq-c-s1/rocketmq471
brokerClusterName = rocketmq-test
namesrvAddr= 172.24.29.215:9976
brokerName = broker-c
#对外提供服务地址
brokerIP1 = 172.24.29.213
#Broker HAIP地址，供slave同步消息的地址
brokerIP2= 172.24.29.213
brokerId = 1
storePathRootDir = /neworiental/rocketmq-test/rocketmq-c-s1/store
storePathCommitLog = /neworiental/rocketmq-test/rocketmq-c-s1/store/commitlog
storePathConsumerQueue = /neworiental/rocketmq-test/rocketmq-c-s1/store/consumequeue
deleteWhen = 04
fileReservedTime = 168
brokerRole = SLAVE
flushDiskType = ASYNC_FLUSH
autoCreateTopicEnable = false
#是否自动创建消费组
autoCreateSubscriptionGroup = false
#集群名称是否可用作Topic使用
clusterTopicEnable = false
#Broker名称是否可用作Topic使用
brokerTopicEnable = false
useEpollNativeSelector = true#是否开启ACL
aclEnable=true#多副本自动主从选举Dledger相关
enableDLegerCommitLog = true
dLegerGroup = broker-c
dLegerSelfId = n1
dLegerPeers = n0-172.24.29.215:10919;n1-172.24.29.213:10919;n2-172.24.29.214:10919

broker-c-s2

listenPort = 20919
rocketmqHome = /neworiental/rocketmq-test/rocketmq-c-s2/rocketmq471
brokerClusterName = rocketmq-test
namesrvAddr= 172.24.29.215:9976
brokerName = broker-c
#对外提供服务地址
brokerIP1 = 172.24.29.214
#Broker HAIP地址，供slave同步消息的地址，不配此项则不会进行主从复制
brokerIP2= 172.24.29.214
brokerId = 2
storePathRootDir = /neworiental/rocketmq-test/rocketmq-c-s2/store
storePathCommitLog = /neworiental/rocketmq-test/rocketmq-c-s2/store/commitlog
storePathConsumerQueue = /neworiental/rocketmq-test/rocketmq-c-s2/store/consumequeue
deleteWhen = 04
fileReservedTime = 168
brokerRole = SLAVE
flushDiskType = ASYNC_FLUSH
autoCreateTopicEnable = false
#是否自动创建消费组
autoCreateSubscriptionGroup = false
#集群名称是否可用作Topic使用
clusterTopicEnable = false
#Broker名称是否可用作Topic使用
brokerTopicEnable = false
useEpollNativeSelector = true#是否开启ACL
aclEnable=true#多副本自动主从选举Dledger相关
enableDLegerCommitLog = true
dLegerGroup = broker-c
dLegerSelfId = n2
dLegerPeers = n0-172.24.29.215:10919;n1-172.24.29.213:10919;n2-172.24.29.214:10919

验证

通过管理命令可以看到集群信息，如下图，每个组有一个Leader，两个Follower，集群搭建成功，日志也正常

Leader日志：

Follower日志：

下面手动将broker-b的Leader杀掉

操作之前：

Leader是在215上

操作之后：

Leader成功选举为214了

再次将215恢复：

可以看到215成为了新的Follower

旧集群升级

如果旧集群采用 Master 方式部署，则每个 Master 都需要转换成一个 RocketMQ-on-DLedger Group。
如果旧集群采用 Master-Slave 方式部署，则每个 Master-Slave 组都需要转换成一个 RocketMQ-on-DLedger Group。

杀掉旧的 Broker

可以通过 kill 命令来完成，也可以调用 bin/mqshutdown broker。

检查旧的 Commitlog

​ RocketMQ-on-DLedger 组中的每个节点，可以兼容旧的 Commitlog ，但其 Raft 复制过程，只能针对新增加的消息。因此，为了避免出现异常，需要保证 旧的 Commitlog 是一致的。
​ 如果旧的集群是采用 Master-Slave 方式部署，有可能在shutdown时，其数据并不是一致的，建议通过md5sum 的方式，检查最近的最少 2 个 Commmitlog 文件，如果发现不一致，则通过拷贝的方式进行对齐。

​ 虽然 RocketMQ-on-DLedger Group 也可以以 2 节点方式部署，但其会丧失容灾切换能力（2n + 1 原则，至少需要3个节点才能容忍其中 1 个宕机）。
​ 所以在对齐了 Master 和 Slave 的 Commitlog 之后，还需要准备第 3 台机器，并把旧的 Commitlog 从 Master 拷贝到 第 3 台机器（记得同时拷贝一下 config 文件夹）。

​ 在 3 台机器准备好了之后，旧 Commitlog 文件也保证一致之后，就可以开始走下一步修改配置了。

修改配置

参考新集群部署。

重新启动 Broker

参考新集群部署。

Tips

  • dLegerPeers：每个DLedger Group内各节点的端口信息，同一个 Group 内的各个节点配置必须要保证一致，多个节点用英文分号隔开，单个条目遵循 legerSlefId-IP:端口，这里的端口用作dledge内部通信，注意不要和RocketMQ本身的端口冲突
  • dLegerSelfId：节点 id, 必须属于dLegerPeers中的一个；同 Group 内各个节点要唯一，并且此ID要和dLegerPeers配置的内容中对应上
  • Broker配置文件中的dLedger少了一个d，不知道是笔误还是有意为之

参考：

https://github.com/apache/rocketmq/blob/master/docs/cn/dledger/deploy_guide.md

这篇关于【RocketMQ】RocketMQ自动容灾切换DLedger集群实践的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/505753。

相关文章

Spring Security简介、使用与最佳实践

Spring Security简介、使用与最佳实践

《SpringSecurity简介、使用与最佳实践》SpringSecurity是一个能够为基于Spring的企业应用系统提供声明式的安全访问控制解决方案的安全框架,本文给大家介绍SpringSec... 目录一、如何理解 Spring Security？—— 核心思想二、如何在 Java 项目中使用？——
阅读更多...
MyBatis Plus实现时间字段自动填充的完整方案

MyBatis Plus实现时间字段自动填充的完整方案

《MyBatisPlus实现时间字段自动填充的完整方案》在日常开发中,我们经常需要记录数据的创建时间和更新时间,传统的做法是在每次插入或更新操作时手动设置这些时间字段,这种方式不仅繁琐,还容易遗漏,... 目录前言解决目标技术栈实现步骤1. 实体类注解配置2. 创建元数据处理器3. 服务层代码优化填充机制详
阅读更多...
防止Linux rm命令误操作的多场景防护方案与实践

防止Linux rm命令误操作的多场景防护方案与实践

《防止Linuxrm命令误操作的多场景防护方案与实践》在Linux系统中,rm命令是删除文件和目录的高效工具,但一旦误操作,如执行rm-rf/或rm-rf/*,极易导致系统数据灾难,本文针对不同场景... 目录引言理解 rm 命令及误操作风险rm 命令基础常见误操作案例防护方案使用 rm编程 别名及安全删除
阅读更多...
C++统计函数执行时间的最佳实践

C++统计函数执行时间的最佳实践

《C++统计函数执行时间的最佳实践》在软件开发过程中,性能分析是优化程序的重要环节,了解函数的执行时间分布对于识别性能瓶颈至关重要,本文将分享一个C++函数执行时间统计工具,希望对大家有所帮助... 目录前言工具特性核心设计1. 数据结构设计2. 单例模式管理器3. RAII自动计时使用方法基本用法高级用法
阅读更多...
PHP应用中处理限流和API节流的最佳实践

PHP应用中处理限流和API节流的最佳实践

《PHP应用中处理限流和API节流的最佳实践》限流和API节流对于确保Web应用程序的可靠性、安全性和可扩展性至关重要,本文将详细介绍PHP应用中处理限流和API节流的最佳实践,下面就来和小编一起学习... 目录限流的重要性在 php 中实施限流的最佳实践使用集中式存储进行状态管理（如 Redis）采用滑动
阅读更多...
ShardingProxy读写分离之原理、配置与实践过程

ShardingProxy读写分离之原理、配置与实践过程

《ShardingProxy读写分离之原理、配置与实践过程》ShardingProxy是ApacheShardingSphere的数据库中间件,通过三层架构实现读写分离,解决高并发场景下数据库性能瓶... 目录一、ShardingProxy技术定位与读写分离核心价值1.1 技术定位1.2 读写分离核心价值二
阅读更多...
深入浅出Spring中的@Autowired自动注入的工作原理及实践应用

深入浅出Spring中的@Autowired自动注入的工作原理及实践应用

《深入浅出Spring中的@Autowired自动注入的工作原理及实践应用》在Spring框架的学习旅程中,@Autowired无疑是一个高频出现却又让初学者头疼的注解,它看似简单,却蕴含着Sprin... 目录深入浅出Spring中的@Autowired：自动注入的奥秘什么是依赖注入？@Autowired
阅读更多...
MySQL分库分表的实践示例

MySQL分库分表的实践示例

《MySQL分库分表的实践示例》MySQL分库分表适用于数据量大或并发压力高的场景,核心技术包括水平/垂直分片和分库,需应对分布式事务、跨库查询等挑战,通过中间件和解决方案实现,最佳实践为合理策略、备... 目录一、分库分表的触发条件1.1 数据量阈值1.2 并发压力二、分库分表的核心技术模块2.1 水平分
阅读更多...
基于Redis自动过期的流处理暂停机制

基于Redis自动过期的流处理暂停机制

《基于Redis自动过期的流处理暂停机制》基于Redis自动过期的流处理暂停机制是一种高效、可靠且易于实现的解决方案,防止延时过大的数据影响实时处理自动恢复处理,以避免积压的数据影响实时性,下面就来详... 目录核心思路代码实现1. 初始化Redis连接和键前缀2. 接收数据时检查暂停状态3. 检测到延时过
阅读更多...
SpringBoot通过main方法启动web项目实践

SpringBoot通过main方法启动web项目实践

《SpringBoot通过main方法启动web项目实践》SpringBoot通过SpringApplication.run()启动Web项目,自动推断应用类型,加载初始化器与监听器,配置Spring... 目录1. 启动入口：SpringApplication.run()2. SpringApplicat
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2