RocketMQ-02

RocketMQ-02 1. RocketMQ集群搭建1.2 集群搭建方式1.2.1 集群特点

推荐整理分享RocketMQ-02，希望有所帮助，仅作参考，欢迎阅读内容。

文章相关热门搜索词:,内容如对您有帮助，希望把文章链接给更多的朋友！

NameServer是一个几乎无状态节点，可集群部署，节点之间无任何信息同步。

Broker部署相对复杂，Broker分为Master与Slave，一个Master可以对应多个Slave，但是一个Slave只能对应一个Master，Master与Slave的对应关系通过指定相同的BrokerName，不同的BrokerId来定义，BrokerId为0表示Master，非0表示Slave。Master也可以部署多个。每个Broker与NameServer集群中的所有节点建立长连接，定时注册Topic信息到所有NameServer。

Producer与NameServer集群中的其中一个节点（随机选择）建立长连接，定期从NameServer取Topic路由信息，并向提供Topic服务的Master建立长连接，且定时向Master发送心跳。Producer完全无状态，可集群部署。

Consumer与NameServer集群中的其中一个节点（随机选择）建立长连接，定期从NameServer取Topic路由信息，并向提供Topic服务的Master、Slave建立长连接，且定时向Master、Slave发送心跳。Consumer既可以从Master订阅消息，也可以从Slave订阅消息，订阅规则由Broker配置决定。

1.2.3 集群模式1）单Master模式

这种方式风险较大，一旦Broker重启或者宕机时，会导致整个服务不可用。不建议线上环境使用,可以用于本地测试。

2）多Master模式

一个集群无Slave，全是Master，例如2个Master或者3个Master，这种模式的优缺点如下：

优点：配置简单，单个Master宕机或重启维护对应用无影响，在磁盘配置为RAID10时，即使机器宕机不可恢复情况下，由于RAID10磁盘非常可靠，消息也不会丢（异步刷盘丢失少量消息，同步刷盘一条不丢），性能最高；缺点：单台机器宕机期间，这台机器上未被消费的消息在机器恢复之前不可订阅，消息实时性会受到影响。3）多Master多Slave模式（异步）

每个Master配置一个Slave，有多对Master-Slave，HA采用异步复制方式，主备有短暂消息延迟（毫秒级），这种模式的优缺点如下：

优点：即使磁盘损坏，消息丢失的非常少，且消息实时性不会受影响，同时Master宕机后，消费者仍然可以从Slave消费，而且此过程对应用透明，不需要人工干预，性能同多Master模式几乎一样；缺点：Master宕机，磁盘损坏情况下会丢失少量消息。4）多Master多Slave模式（同步）

每个Master配置一个Slave，有多对Master-Slave，HA采用同步双写方式，即只有主备都写成功，才向应用返回成功，这种模式的优缺点如下：

优点：数据与服务都无单点故障，Master宕机情况下，消息无延迟，服务可用性与数据可用性都非常高；缺点：性能比异步复制模式略低（大约低10%左右），发送单个消息的RT会略高，且目前版本在主节点宕机后，备机不能自动切换为主机。1.3 双主双从集群搭建1.3.1 总体架构

消息高可用采用2m-2s（同步双写）方式

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-cICQckhC-1681918953093)(assets/RocketMQ集群.png)]

1.3.2 集群工作流程启动NameServer，NameServer起来后监听端口，等待Broker、Producer、Consumer连上来，相当于一个路由控制中心。Broker启动，跟所有的NameServer保持长连接，定时发送心跳包。心跳包中包含当前Broker信息(IP+端口等)以及存储所有Topic信息。注册成功后，NameServer集群中就有Topic跟Broker的映射关系。收发消息前，先创建Topic，创建Topic时需要指定该Topic要存储在哪些Broker上，也可以在发送消息时自动创建Topic。Producer发送消息，启动时先跟NameServer集群中的其中一台建立长连接，并从NameServer中获取当前发送的Topic存在哪些Broker上，轮询从队列列表中选择一个队列，然后与队列所在的Broker建立长连接从而向Broker发消息。Consumer跟Producer类似，跟其中一台NameServer建立长连接，获取当前订阅Topic存在哪些Broker上，然后直接跟Broker建立连接通道，开始消费消息。1.3.3 服务器环境序号IP角色架构模式1192.168.25.135nameserver、brokerserverMaster1、Slave22192.168.25.138nameserver、brokerserverMaster2、Slave13.3.4 Host添加信息vim /etc/hosts

配置如下:

# nameserver192.168.25.135 rocketmq-nameserver1192.168.25.138 rocketmq-nameserver2# broker192.168.25.135 rocketmq-master1192.168.25.135 rocketmq-slave2192.168.25.138 rocketmq-master2192.168.25.138 rocketmq-slave1# 本人设置，已更改ip地址# nameserver192.168.200.128 rocketmq-nameserver1192.168.200.129 rocketmq-nameserver2# broker 192.168.200.128 rocketmq-master1192.168.200.128 rocketmq-slave2192.168.200.129 rocketmq-master2192.168.200.129 rocketmq-slave1

配置完成后, 重启网卡

systemctl restart network1.3.5 防火墙配置

宿主机需要远程访问虚拟机的rocketmq服务和web服务，需要开放相关的端口号，简单粗暴的方式是直接关闭防火墙

# 关闭防火墙systemctl stop firewalld.service # 查看防火墙的状态firewall-cmd --state # 禁止firewall开机启动systemctl disable firewalld.service

或者为了安全，只开放特定的端口号，RocketMQ默认使用3个端口：9876 、10911 、11011 。如果防火墙没有关闭的话，那么防火墙就必须开放这些端口：

nameserver 默认使用 9876 端口master 默认使用 10911 端口slave 默认使用11011 端口

执行以下命令：

# 开放name server默认端口firewall-cmd --remove-port=9876/tcp --permanent# 开放master默认端口firewall-cmd --remove-port=10911/tcp --permanent# 开放slave默认端口 (当前集群模式可不开启)firewall-cmd --remove-port=11011/tcp --permanent # 重启防火墙firewall-cmd --reload1.3.6 环境变量配置vim /etc/profile

在profile文件的末尾加入如下命令

#set rocketmqROCKETMQ_HOME=/usr/local/rocketmq/rocketmq-all-4.4.0-bin-releasePATH=$PATH:$ROCKETMQ_HOME/binexport ROCKETMQ_HOME PATH#set rocketmq 本人设置，已更改 rocketmq 安装目录ROCKETMQ_HOME=/opt/rocketmqPATH=$PATH:$ROCKETMQ_HOME/binexport ROCKETMQ_HOME PATH

输入:wq! 保存并退出， 并使得配置立刻生效：

source /etc/profile1.3.7 创建消息存储路径mkdir /usr/local/rocketmq/storemkdir /usr/local/rocketmq/store/commitlogmkdir /usr/local/rocketmq/store/consumequeuemkdir /usr/local/rocketmq/store/index本人自己设置目录 主目录mkdir /opt/rocketmq-store/storemkdir /opt/rocketmq-store/store/commitlogmkdir /opt/rocketmq-store/store/consumequeuemkdir /opt/rocketmq-store/store/index本人自己设置目录 从目录mkdir /opt/rocketmq-store-slave/storemkdir /opt/rocketmq-store-slave/store/commitlogmkdir /opt/rocketmq-store-slave/store/consumequeuemkdir /opt/rocketmq-store-slave/store/index1.3.8 broker配置文件1）master1

服务器：192.168.25.135

vi /usr/soft/rocketmq/conf/2m-2s-sync/broker-a.properties本人设置 目录vi /opt/rocketmq/conf/2m-2s-sync/broker-a.properties

修改配置如下：

#所属集群名字brokerClusterName=rocketmq-cluster#broker名字，注意此处不同的配置文件填写的不一样brokerName=broker-a#0 表示 Master，>0 表示 SlavebrokerId=0#nameServer地址，分号分割namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2:9876#在发送消息时，自动创建服务器不存在的topic，默认创建的队列数defaultTopicQueueNums=4#是否允许 Broker 自动创建Topic，建议线下开启，线上关闭autoCreateTopicEnable=true#是否允许 Broker 自动创建订阅组，建议线下开启，线上关闭autoCreateSubscriptionGroup=true#Broker 对外服务的监听端口listenPort=10911#删除文件时间点，默认凌晨 4点deleteWhen=04#文件保留时间，默认 48 小时fileReservedTime=120#commitLog每个文件的大小默认1GmapedFileSizeCommitLog=1073741824#ConsumeQueue每个文件默认存30W条，根据业务情况调整mapedFileSizeConsumeQueue=300000#destroyMapedFileIntervalForcibly=120000#redeleteHangedFileInterval=120000#检测物理文件磁盘空间diskMaxUsedSpaceRatio=88#存储路径storePathRootDir=/usr/local/rocketmq/store#commitLog 存储路径storePathCommitLog=/usr/local/rocketmq/store/commitlog#消费队列存储路径存储路径storePathConsumeQueue=/usr/local/rocketmq/store/consumequeue#消息索引存储路径storePathIndex=/usr/local/rocketmq/store/index#checkpoint 文件存储路径storeCheckpoint=/usr/local/rocketmq/store/checkpoint#abort 文件存储路径abortFile=/usr/local/rocketmq/store/abort#限制的消息大小maxMessageSize=65536#flushCommitLogLeastPages=4#flushConsumeQueueLeastPages=2#flushCommitLogThoroughInterval=10000#flushConsumeQueueThoroughInterval=60000#Broker 的角色#- ASYNC_MASTER 异步复制Master#- SYNC_MASTER 同步双写Master#- SLAVEbrokerRole=SYNC_MASTER#刷盘方式#- ASYNC_FLUSH 异步刷盘#- SYNC_FLUSH 同步刷盘flushDiskType=SYNC_FLUSH#checkTransactionMessageEnable=false#发消息线程池数量#sendMessageThreadPoolNums=128#拉消息线程池数量#pullMessageThreadPoolNums=128服务器：192.168.200.128设置的位置目录修改#所属集群名字brokerClusterName=rocketmq-cluster#broker名字，注意此处不同的配置文件填写的不一样brokerName=broker-a#0 表示 Master，>0 表示 SlavebrokerId=0#nameServer地址，分号分割namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2:9876#在发送消息时，自动创建服务器不存在的topic，默认创建的队列数defaultTopicQueueNums=4#是否允许 Broker 自动创建Topic，建议线下开启，线上关闭autoCreateTopicEnable=true#是否允许 Broker 自动创建订阅组，建议线下开启，线上关闭autoCreateSubscriptionGroup=true#Broker 对外服务的监听端口listenPort=10911#删除文件时间点，默认凌晨 4点deleteWhen=04#文件保留时间，默认 48 小时fileReservedTime=120#commitLog每个文件的大小默认1GmapedFileSizeCommitLog=1073741824#ConsumeQueue每个文件默认存30W条，根据业务情况调整mapedFileSizeConsumeQueue=300000#destroyMapedFileIntervalForcibly=120000#redeleteHangedFileInterval=120000#检测物理文件磁盘空间diskMaxUsedSpaceRatio=88#存储路径storePathRootDir=/opt/rocketmq-store/store#commitLog 存储路径storePathCommitLog=/opt/rocketmq-store/store/commitlog#消费队列存储路径存储路径storePathConsumeQueue=/opt/rocketmq-store/store/consumequeue#消息索引存储路径storePathIndex=/opt/rocketmq-store/store/index#checkpoint 文件存储路径storeCheckpoint=/opt/rocketmq-store/store/checkpoint#abort 文件存储路径abortFile=/opt/rocketmq-store/store/abort#限制的消息大小maxMessageSize=65536#flushCommitLogLeastPages=4#flushConsumeQueueLeastPages=2#flushCommitLogThoroughInterval=10000#flushConsumeQueueThoroughInterval=60000#Broker 的角色#- ASYNC_MASTER 异步复制Master#- SYNC_MASTER 同步双写Master#- SLAVEbrokerRole=SYNC_MASTER#刷盘方式#- ASYNC_FLUSH 异步刷盘#- SYNC_FLUSH 同步刷盘flushDiskType=SYNC_FLUSH#checkTransactionMessageEnable=false#发消息线程池数量#sendMessageThreadPoolNums=128#拉消息线程池数量#pullMessageThreadPoolNums=1282）slave2

服务器：192.168.25.135

vi /usr/soft/rocketmq/conf/2m-2s-sync/broker-b-s.properties本人设置 目录vi /opt/rocketmq/conf/2m-2s-sync/broker-b-s.properties

修改配置如下：

#所属集群名字brokerClusterName=rocketmq-cluster#broker名字，注意此处不同的配置文件填写的不一样brokerName=broker-b#0 表示 Master，>0 表示 SlavebrokerId=1#nameServer地址，分号分割namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2:9876#在发送消息时，自动创建服务器不存在的topic，默认创建的队列数defaultTopicQueueNums=4#是否允许 Broker 自动创建Topic，建议线下开启，线上关闭autoCreateTopicEnable=true#是否允许 Broker 自动创建订阅组，建议线下开启，线上关闭autoCreateSubscriptionGroup=true#Broker 对外服务的监听端口listenPort=11011#删除文件时间点，默认凌晨 4点deleteWhen=04#文件保留时间，默认 48 小时fileReservedTime=120#commitLog每个文件的大小默认1GmapedFileSizeCommitLog=1073741824#ConsumeQueue每个文件默认存30W条，根据业务情况调整mapedFileSizeConsumeQueue=300000#destroyMapedFileIntervalForcibly=120000#redeleteHangedFileInterval=120000#检测物理文件磁盘空间diskMaxUsedSpaceRatio=88#存储路径storePathRootDir=/usr/local/rocketmq/store-slave#commitLog 存储路径storePathCommitLog=/usr/local/rocketmq/store/commitlog#消费队列存储路径存储路径storePathConsumeQueue=/usr/local/rocketmq/store/consumequeue#消息索引存储路径storePathIndex=/usr/local/rocketmq/store/index#checkpoint 文件存储路径storeCheckpoint=/usr/local/rocketmq/store/checkpoint#abort 文件存储路径abortFile=/usr/local/rocketmq/store/abort#限制的消息大小maxMessageSize=65536#flushCommitLogLeastPages=4#flushConsumeQueueLeastPages=2#flushCommitLogThoroughInterval=10000#flushConsumeQueueThoroughInterval=60000#Broker 的角色#- ASYNC_MASTER 异步复制Master#- SYNC_MASTER 同步双写Master#- SLAVEbrokerRole=SLAVE#刷盘方式#- ASYNC_FLUSH 异步刷盘#- SYNC_FLUSH 同步刷盘flushDiskType=ASYNC_FLUSH#checkTransactionMessageEnable=false#发消息线程池数量#sendMessageThreadPoolNums=128#拉消息线程池数量#pullMessageThreadPoolNums=128服务器：192.168.200.128本人设置，位置已更改#所属集群名字brokerClusterName=rocketmq-cluster#broker名字，注意此处不同的配置文件填写的不一样brokerName=broker-b#0 表示 Master，>0 表示 SlavebrokerId=1#nameServer地址，分号分割namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2:9876#在发送消息时，自动创建服务器不存在的topic，默认创建的队列数defaultTopicQueueNums=4#是否允许 Broker 自动创建Topic，建议线下开启，线上关闭autoCreateTopicEnable=true#是否允许 Broker 自动创建订阅组，建议线下开启，线上关闭autoCreateSubscriptionGroup=true#Broker 对外服务的监听端口listenPort=11011#删除文件时间点，默认凌晨 4点deleteWhen=04#文件保留时间，默认 48 小时fileReservedTime=120#commitLog每个文件的大小默认1GmapedFileSizeCommitLog=1073741824#ConsumeQueue每个文件默认存30W条，根据业务情况调整mapedFileSizeConsumeQueue=300000#destroyMapedFileIntervalForcibly=120000#redeleteHangedFileInterval=120000#检测物理文件磁盘空间diskMaxUsedSpaceRatio=88#存储路径storePathRootDir=/opt/rocketmq-store/store#commitLog 存储路径storePathCommitLog=/opt/rocketmq-store/store/commitlog#消费队列存储路径存储路径storePathConsumeQueue=/opt/rocketmq-store/store/consumequeue#消息索引存储路径storePathIndex=/opt/rocketmq-store/store/index#checkpoint 文件存储路径storeCheckpoint=/opt/rocketmq-store/store/checkpoint#abort 文件存储路径abortFile=/opt/rocketmq-store/store/abort#限制的消息大小maxMessageSize=65536#flushCommitLogLeastPages=4#flushConsumeQueueLeastPages=2#flushCommitLogThoroughInterval=10000#flushConsumeQueueThoroughInterval=60000#Broker 的角色#- ASYNC_MASTER 异步复制Master#- SYNC_MASTER 同步双写Master#- SLAVEbrokerRole=SLAVE#刷盘方式#- ASYNC_FLUSH 异步刷盘#- SYNC_FLUSH 同步刷盘flushDiskType=ASYNC_FLUSH#checkTransactionMessageEnable=false#发消息线程池数量#sendMessageThreadPoolNums=128#拉消息线程池数量#pullMessageThreadPoolNums=1283）master2

服务器：192.168.25.138 192.168.200.129

vi /usr/soft/rocketmq/conf/2m-2s-sync/broker-b.properties本人设置 目录vi /opt/rocketmq/conf/2m-2s-sync/broker-b.properties

修改配置如下：

#所属集群名字brokerClusterName=rocketmq-cluster#broker名字，注意此处不同的配置文件填写的不一样brokerName=broker-b#0 表示 Master，>0 表示 SlavebrokerId=0#nameServer地址，分号分割namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2:9876#在发送消息时，自动创建服务器不存在的topic，默认创建的队列数defaultTopicQueueNums=4#是否允许 Broker 自动创建Topic，建议线下开启，线上关闭autoCreateTopicEnable=true#是否允许 Broker 自动创建订阅组，建议线下开启，线上关闭autoCreateSubscriptionGroup=true#Broker 对外服务的监听端口listenPort=10911#删除文件时间点，默认凌晨 4点deleteWhen=04#文件保留时间，默认 48 小时fileReservedTime=120#commitLog每个文件的大小默认1GmapedFileSizeCommitLog=1073741824#ConsumeQueue每个文件默认存30W条，根据业务情况调整mapedFileSizeConsumeQueue=300000#destroyMapedFileIntervalForcibly=120000#redeleteHangedFileInterval=120000#检测物理文件磁盘空间diskMaxUsedSpaceRatio=88#存储路径storePathRootDir=/usr/local/rocketmq/store#commitLog 存储路径storePathCommitLog=/usr/local/rocketmq/store/commitlog#消费队列存储路径存储路径storePathConsumeQueue=/usr/local/rocketmq/store/consumequeue#消息索引存储路径storePathIndex=/usr/local/rocketmq/store/index#checkpoint 文件存储路径storeCheckpoint=/usr/local/rocketmq/store/checkpoint#abort 文件存储路径abortFile=/usr/local/rocketmq/store/abort#限制的消息大小maxMessageSize=65536#flushCommitLogLeastPages=4#flushConsumeQueueLeastPages=2#flushCommitLogThoroughInterval=10000#flushConsumeQueueThoroughInterval=60000#Broker 的角色#- ASYNC_MASTER 异步复制Master#- SYNC_MASTER 同步双写Master#- SLAVEbrokerRole=SYNC_MASTER#刷盘方式#- ASYNC_FLUSH 异步刷盘#- SYNC_FLUSH 同步刷盘flushDiskType=SYNC_FLUSH#checkTransactionMessageEnable=false#发消息线程池数量#sendMessageThreadPoolNums=128#拉消息线程池数量#pullMessageThreadPoolNums=128服务器：192.168.200.129本人设置，位置已更改#所属集群名字brokerClusterName=rocketmq-cluster#broker名字，注意此处不同的配置文件填写的不一样brokerName=broker-b#0 表示 Master，>0 表示 SlavebrokerId=0#nameServer地址，分号分割namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2:9876#在发送消息时，自动创建服务器不存在的topic，默认创建的队列数defaultTopicQueueNums=4#是否允许 Broker 自动创建Topic，建议线下开启，线上关闭autoCreateTopicEnable=true#是否允许 Broker 自动创建订阅组，建议线下开启，线上关闭autoCreateSubscriptionGroup=true#Broker 对外服务的监听端口listenPort=10911#删除文件时间点，默认凌晨 4点deleteWhen=04#文件保留时间，默认 48 小时fileReservedTime=120#commitLog每个文件的大小默认1GmapedFileSizeCommitLog=1073741824#ConsumeQueue每个文件默认存30W条，根据业务情况调整mapedFileSizeConsumeQueue=300000#destroyMapedFileIntervalForcibly=120000#redeleteHangedFileInterval=120000#检测物理文件磁盘空间diskMaxUsedSpaceRatio=88#存储路径storePathRootDir=/opt/rocketmq-store/store#commitLog 存储路径storePathCommitLog=/opt/rocketmq-store/store/commitlog#消费队列存储路径存储路径storePathConsumeQueue=/opt/rocketmq-store/store/consumequeue#消息索引存储路径storePathIndex=/opt/rocketmq-store/store/index#checkpoint 文件存储路径storeCheckpoint=/opt/rocketmq-store/store/checkpoint#abort 文件存储路径abortFile=/opt/rocketmq-store/store/abort#限制的消息大小maxMessageSize=65536#flushCommitLogLeastPages=4#flushConsumeQueueLeastPages=2#flushCommitLogThoroughInterval=10000#flushConsumeQueueThoroughInterval=60000#Broker 的角色#- ASYNC_MASTER 异步复制Master#- SYNC_MASTER 同步双写Master#- SLAVEbrokerRole=SYNC_MASTER#刷盘方式#- ASYNC_FLUSH 异步刷盘#- SYNC_FLUSH 同步刷盘flushDiskType=SYNC_FLUSH#checkTransactionMessageEnable=false#发消息线程池数量#sendMessageThreadPoolNums=128#拉消息线程池数量#pullMessageThreadPoolNums=1284）slave1

服务器：192.168.25.138 192.168.200.129

vi /usr/soft/rocketmq/conf/2m-2s-sync/broker-a-s.properties本人设置 目录vi /opt/rocketmq/conf/2m-2s-sync/broker-a-s.properties

修改配置如下：

#所属集群名字brokerClusterName=rocketmq-cluster#broker名字，注意此处不同的配置文件填写的不一样brokerName=broker-a#0 表示 Master，>0 表示 SlavebrokerId=1#nameServer地址，分号分割namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2:9876#在发送消息时，自动创建服务器不存在的topic，默认创建的队列数defaultTopicQueueNums=4#是否允许 Broker 自动创建Topic，建议线下开启，线上关闭autoCreateTopicEnable=true#是否允许 Broker 自动创建订阅组，建议线下开启，线上关闭autoCreateSubscriptionGroup=true#Broker 对外服务的监听端口listenPort=11011#删除文件时间点，默认凌晨 4点deleteWhen=04#文件保留时间，默认 48 小时fileReservedTime=120#commitLog每个文件的大小默认1GmapedFileSizeCommitLog=1073741824#ConsumeQueue每个文件默认存30W条，根据业务情况调整mapedFileSizeConsumeQueue=300000#destroyMapedFileIntervalForcibly=120000#redeleteHangedFileInterval=120000#检测物理文件磁盘空间diskMaxUsedSpaceRatio=88#存储路径storePathRootDir=/usr/local/rocketmq/store#commitLog 存储路径storePathCommitLog=/usr/local/rocketmq/store/commitlog#消费队列存储路径存储路径storePathConsumeQueue=/usr/local/rocketmq/store/consumequeue#消息索引存储路径storePathIndex=/usr/local/rocketmq/store/index#checkpoint 文件存储路径storeCheckpoint=/usr/local/rocketmq/store/checkpoint#abort 文件存储路径abortFile=/usr/local/rocketmq/store/abort#限制的消息大小maxMessageSize=65536#flushCommitLogLeastPages=4#flushConsumeQueueLeastPages=2#flushCommitLogThoroughInterval=10000#flushConsumeQueueThoroughInterval=60000#Broker 的角色#- ASYNC_MASTER 异步复制Master#- SYNC_MASTER 同步双写Master#- SLAVEbrokerRole=SLAVE#刷盘方式#- ASYNC_FLUSH 异步刷盘#- SYNC_FLUSH 同步刷盘flushDiskType=ASYNC_FLUSH#checkTransactionMessageEnable=false#发消息线程池数量#sendMessageThreadPoolNums=128#拉消息线程池数量#pullMessageThreadPoolNums=128服务器：192.168.200.129本人设置，位置已更改#所属集群名字brokerClusterName=rocketmq-cluster#broker名字，注意此处不同的配置文件填写的不一样brokerName=broker-a#0 表示 Master，>0 表示 SlavebrokerId=1#nameServer地址，分号分割namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2:9876#在发送消息时，自动创建服务器不存在的topic，默认创建的队列数defaultTopicQueueNums=4#是否允许 Broker 自动创建Topic，建议线下开启，线上关闭autoCreateTopicEnable=true#是否允许 Broker 自动创建订阅组，建议线下开启，线上关闭autoCreateSubscriptionGroup=true#Broker 对外服务的监听端口listenPort=11011#删除文件时间点，默认凌晨 4点deleteWhen=04#文件保留时间，默认 48 小时fileReservedTime=120#commitLog每个文件的大小默认1GmapedFileSizeCommitLog=1073741824#ConsumeQueue每个文件默认存30W条，根据业务情况调整mapedFileSizeConsumeQueue=300000#destroyMapedFileIntervalForcibly=120000#redeleteHangedFileInterval=120000#检测物理文件磁盘空间diskMaxUsedSpaceRatio=88#存储路径storePathRootDir=/opt/rocketmq-store/store#commitLog 存储路径storePathCommitLog=/opt/rocketmq-store/store/commitlog#消费队列存储路径存储路径storePathConsumeQueue=/opt/rocketmq-store/store/consumequeue#消息索引存储路径storePathIndex=/opt/rocketmq-store/store/index#checkpoint 文件存储路径storeCheckpoint=/opt/rocketmq-store/store/checkpoint#abort 文件存储路径abortFile=/opt/rocketmq-store/store/abort#限制的消息大小maxMessageSize=65536#flushCommitLogLeastPages=4#flushConsumeQueueLeastPages=2#flushCommitLogThoroughInterval=10000#flushConsumeQueueThoroughInterval=60000#Broker 的角色#- ASYNC_MASTER 异步复制Master#- SYNC_MASTER 同步双写Master#- SLAVEbrokerRole=SLAVE#刷盘方式#- ASYNC_FLUSH 异步刷盘#- SYNC_FLUSH 同步刷盘flushDiskType=ASYNC_FLUSH#checkTransactionMessageEnable=false#发消息线程池数量#sendMessageThreadPoolNums=128#拉消息线程池数量#pullMessageThreadPoolNums=1281.3.9 修改启动脚本文件1）runbroker.shvi /usr/local/rocketmq/bin/runbroker.sh本人设置vi /opt/rocketmq/bin/runbroker.sh

需要根据内存大小进行适当的对JVM参数进行调整：

#===================================================# 开发环境配置 JVM ConfigurationJAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m"

####2）runserver.sh

vim /usr/local/rocketmq/bin/runserver.sh本人设置vi /opt/rocketmq/bin/runbroker.shJAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"1.3.10 服务启动1）启动NameServe集群

分别在192.168.25.135和192.168.25.138启动NameServer

cd /usr/local/rocketmq/binnohup sh mqnamesrv &2）启动Broker集群在192.168.25.135上启动master1和slave2

master1：

cd /usr/local/rocketmq/binnohup sh mqbroker -c /usr/local/rocketmq/conf/2m-2s-sync/broker-a.properties &

slave2：

cd /usr/local/rocketmq/binnohup sh mqbroker -c /usr/local/rocketmq/conf/2m-2s-sync/broker-b-s.properties &在192.168.25.138上启动master2和slave2

master2

cd /usr/local/rocketmq/binnohup sh mqbroker -c /usr/local/rocketmq/conf/2m-2s-sync/broker-b.properties &

slave1

cd /usr/local/rocketmq/binnohup sh mqbroker -c /usr/local/rocketmq/conf/2m-2s-sync/broker-a-s.properties &1.3.11 查看进程状态

启动后通过JPS查看启动进程

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-t4TwJLkE-1681918953098)(assets/jps1.png)]

1.3.12 查看日志# 查看nameServer日志tail -500f ~/logs/rocketmqlogs/namesrv.log# 查看broker日志tail -500f ~/logs/rocketmqlogs/broker.log3.4 mqadmin管理工具1.4.1 使用方式

进入RocketMQ安装位置，在bin目录下执行./mqadmin {command} {args}

###1.4.2 命令介绍

####1）Topic相关

名称含义命令选项说明updateTopic创建更新Topic配置-bBroker 地址，表示 topic 所在 Broker，只支持单台Broker，地址为ip:port-ccluster 名称，表示 topic 所在集群（集群可通过 clusterList 查询）-h-打印帮助-nNameServer服务地址，格式 ip:port-p指定新topic的读写权限( W=2|R=4|WR=6 )-r可读队列数（默认为 8）-w可写队列数（默认为 8）-ttopic 名称（名称只能使用字符 ^[a-zA-Z0-9_-]+$ ）deleteTopic删除Topic-ccluster 名称，表示删除某集群下的某个 topic （集群 可通过 clusterList 查询）-h打印帮助-nNameServer 服务地址，格式 ip:port-ttopic 名称（名称只能使用字符 ^[a-zA-Z0-9_-]+$ ）topicList查看 Topic 列表信息-h打印帮助-c不配置-c只返回topic列表，增加-c返回clusterName, topic, consumerGroup信息，即topic的所属集群和订阅关系，没有参数-nNameServer 服务地址，格式 ip:porttopicRoute查看 Topic 路由信息-ttopic 名称-h打印帮助-nNameServer 服务地址，格式 ip:porttopicStatus查看 Topic 消息队列offset-ttopic 名称-h打印帮助-nNameServer 服务地址，格式 ip:porttopicClusterList查看 Topic 所在集群列表-ttopic 名称-h打印帮助-nNameServer 服务地址，格式 ip:portupdateTopicPerm更新 Topic 读写权限-ttopic 名称-h打印帮助-nNameServer 服务地址，格式 ip:port-bBroker 地址，表示 topic 所在 Broker，只支持单台Broker，地址为ip:port-p指定新 topic 的读写权限( W=2|R=4|WR=6 )-ccluster 名称，表示 topic 所在集群（集群可通过 clusterList 查询），-b优先，如果没有-b，则对集群中所有Broker执行命令updateOrderConf从NameServer上创建、删除、获取特定命名空间的kv配置，目前还未启用-h打印帮助-nNameServer 服务地址，格式 ip:port-ttopic，键-vorderConf，值-mmethod，可选get、put、deleteallocateMQ以平均负载算法计算消费者列表负载消息队列的负载结果-ttopic 名称-h打印帮助-nNameServer 服务地址，格式 ip:port-iipList，用逗号分隔，计算这些ip去负载Topic的消息队列statsAll打印Topic订阅关系、TPS、积累量、24h读写总量等信息-h打印帮助-nNameServer 服务地址，格式 ip:port-a是否只打印活跃topic-t指定topic

####2）集群相关

名称含义命令选项说明 clusterList查看集群信息，集群、BrokerName、BrokerId、TPS等信息-m打印更多信息 (增加打印出如下信息 #InTotalYest, #OutTotalYest, #InTotalToday ,#OutTotalToday)-h打印帮助-nNameServer 服务地址，格式 ip:port-i打印间隔，单位秒clusterRT发送消息检测集群各Broker RT。消息发往${BrokerName} Topic。-aamount，每次探测的总数，RT = 总时间 / amount-s消息大小，单位B-c探测哪个集群-p是否打印格式化日志，以|分割，默认不打印-h打印帮助-m所属机房，打印使用-i发送间隔，单位秒-nNameServer 服务地址，格式 ip:port

####3）Broker相关

名称含义命令选项说明updateBrokerConfig更新 Broker 配置文件，会修改Broker.conf-bBroker 地址，格式为ip:port-ccluster 名称-kkey 值-vvalue 值-h打印帮助-nNameServer 服务地址，格式 ip:portbrokerStatus查看 Broker 统计信息、运行状态（你想要的信息几乎都在里面）-bBroker 地址，地址为ip:port-h打印帮助-nNameServer 服务地址，格式 ip:portbrokerConsumeStatsBroker中各个消费者的消费情况，按Message Queue维度返回Consume Offset，Broker Offset，Diff，TImestamp等信息-bBroker 地址，地址为ip:port-t请求超时时间-ldiff阈值，超过阈值才打印-o是否为顺序topic，一般为false-h打印帮助-nNameServer 服务地址，格式 ip:portgetBrokerConfig获取Broker配置-bBroker 地址，地址为ip:port-nNameServer 服务地址，格式 ip:portwipeWritePerm从NameServer上清除 Broker写权限-bBroker 地址，地址为ip:port-nNameServer 服务地址，格式 ip:port-h打印帮助cleanExpiredCQ清理Broker上过期的Consume Queue，如果手动减少对列数可能产生过期队列-nNameServer 服务地址，格式 ip:port-h打印帮助-bBroker 地址，地址为ip:port-c集群名称cleanUnusedTopic清理Broker上不使用的Topic，从内存中释放Topic的Consume Queue，如果手动删除Topic会产生不使用的Topic-nNameServer 服务地址，格式 ip:port-h打印帮助-bBroker 地址，地址为ip:port-c集群名称sendMsgStatus向Broker发消息，返回发送状态和RT-nNameServer 服务地址，格式 ip:port-h打印帮助-bBrokerName，注意不同于Broker地址-s消息大小，单位B-c发送次数

####4）消息相关

名称含义命令选项说明queryMsgById根据offsetMsgId查询msg，如果使用开源控制台，应使用offsetMsgId，此命令还有其他参数，具体作用请阅读QueryMsgByIdSubCommand。-imsgId-h打印帮助-nNameServer 服务地址，格式 ip:portqueryMsgByKey根据消息 Key 查询消息-kmsgKey-tTopic 名称-h打印帮助-nNameServer 服务地址，格式 ip:portqueryMsgByOffset根据 Offset 查询消息-bBroker 名称，（这里需要注意 填写的是 Broker 的名称，不是 Broker 的地址，Broker 名称可以在 clusterList 查到）-iquery 队列 id-ooffset 值-ttopic 名称-h打印帮助-nNameServer 服务地址，格式 ip:portqueryMsgByUniqueKey根据msgId查询，msgId不同于offsetMsgId，区别详见常见运维问题。-g，-d配合使用，查到消息后尝试让特定的消费者消费消息并返回消费结果-h打印帮助-nNameServer 服务地址，格式 ip:port-iuniqe msg id-gconsumerGroup-dclientId-ttopic名称checkMsgSendRT检测向topic发消息的RT，功能类似clusterRT-h打印帮助-nNameServer 服务地址，格式 ip:port-ttopic名称-a探测次数-s消息大小sendMessage发送一条消息，可以根据配置发往特定Message Queue，或普通发送。-h打印帮助-nNameServer 服务地址，格式 ip:port-ttopic名称-pbody，消息体-kkeys-ctags-bBrokerName-iqueueIdconsumeMessage消费消息。可以根据offset、开始&结束时间戳、消息队列消费消息，配置不同执行不同消费逻辑，详见ConsumeMessageCommand。-h打印帮助-nNameServer 服务地址，格式 ip:port-ttopic名称-bBrokerName-o从offset开始消费-iqueueId-g消费者分组-s开始时间戳，格式详见-h-d结束时间戳-c消费多少条消息printMsg从Broker消费消息并打印，可选时间段-h打印帮助-nNameServer 服务地址，格式 ip:port-ttopic名称-c字符集，例如UTF-8-ssubExpress，过滤表达式-b开始时间戳，格式参见-h-e结束时间戳-d是否打印消息体printMsgByQueue类似printMsg，但指定Message Queue-h打印帮助-nNameServer 服务地址，格式 ip:port-ttopic名称-iqueueId-aBrokerName-c字符集，例如UTF-8-ssubExpress，过滤表达式-b开始时间戳，格式参见-h-e结束时间戳-p是否打印消息-d是否打印消息体-f是否统计tag数量并打印resetOffsetByTime按时间戳重置offset，Broker和consumer都会重置-h打印帮助-nNameServer 服务地址，格式 ip:port-g消费者分组-ttopic名称-s重置为此时间戳对应的offset-f是否强制重置，如果false，只支持回溯offset，如果true，不管时间戳对应offset与consumeOffset关系-c是否重置c++客户端offset5）消费者、消费组相关名称含义命令选项说明consumerProgress查看订阅组消费状态，可以查看具体的client IP的消息积累量-g消费者所属组名-s是否打印client IP-h打印帮助-nNameServer 服务地址，格式 ip:portconsumerStatus查看消费者状态，包括同一个分组中是否都是相同的订阅，分析Process Queue是否堆积，返回消费者jstack结果，内容较多，使用者参见ConsumerStatusSubCommand-h打印帮助-nNameServer 服务地址，格式 ip:port-gconsumer group-iclientId-s是否执行jstackgetConsumerStatus获取 Consumer 消费进度-g消费者所属组名-t查询主题-iConsumer 客户端 ip-nNameServer 服务地址，格式 ip:port-h打印帮助updateSubGroup更新或创建订阅关系-nNameServer 服务地址，格式 ip:port-h打印帮助-bBroker地址-c集群名称-g消费者分组名称-s分组是否允许消费-m是否从最小offset开始消费-d是否是广播模式-q重试队列数量-r最大重试次数-i当slaveReadEnable开启时有效，且还未达到从slave消费时建议从哪个BrokerId消费，可以配置备机id，主动从备机消费-w如果Broker建议从slave消费，配置决定从哪个slave消费，配置BrokerId，例如1-a当消费者数量变化时是否通知其他消费者负载均衡deleteSubGroup从Broker删除订阅关系-nNameServer 服务地址，格式 ip:port-h打印帮助-bBroker地址-c集群名称-g消费者分组名称cloneGroupOffset在目标群组中使用源群组的offset-nNameServer 服务地址，格式 ip:port-h打印帮助-s源消费者组-d目标消费者组-ttopic名称-o暂未使用6）连接相关名称含义命令选项说明consumerConnec tion查询 Consumer 的网络连接-g消费者所属组名-nNameServer 服务地址，格式 ip:port-h打印帮助producerConnec tion查询 Producer 的网络连接-g生产者所属组名-t主题名称-nNameServer 服务地址，格式 ip:port-h打印帮助7）NameServer相关名称含义命令选项说明updateKvConfig更新NameServer的kv配置，目前还未使用-s命名空间-kkey-vvalue-nNameServer 服务地址，格式 ip:port-h打印帮助deleteKvConfig删除NameServer的kv配置-s命名空间-kkey-nNameServer 服务地址，格式 ip:port-h打印帮助getNamesrvConfig获取NameServer配置-nNameServer 服务地址，格式 ip:port-h打印帮助updateNamesrvConfig修改NameServer配置-nNameServer 服务地址，格式 ip:port-h打印帮助-kkey-vvalue8）其他名称含义命令选项说明startMonitoring开启监控进程，监控消息误删、重试队列消息数等-nNameServer 服务地址，格式 ip:port-h打印帮助1.4.3 注意事项几乎所有命令都需要配置-n表示NameServer地址，格式为ip:port几乎所有命令都可以通过-h获取帮助如果既有Broker地址（-b）配置项又有clusterName（-c）配置项，则优先以Broker地址执行命令；如果不配置Broker地址，则对集群中所有主机执行命令3.5 集群监控平台搭建1.5.1 概述

RocketMQ有一个对其扩展的开源项目incubator-rocketmq-externals，这个项目中有一个子模块叫rocketmq-console，这个便是管理控制台项目了，先将incubator-rocketmq-externals拉到本地，因为我们需要自己对rocketmq-console进行编译打包运行。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-TBAUKn8S-1681918953101)(assets/rocketmq-console.png)]

1.5.2 下载并编译打包git clone https://github.com/apache/rocketmq-externalscd rocketmq-consolemvn clean package -Dmaven.test.skip=true

注意：打包前在rocketmq-console中配置namesrv集群地址：

rocketmq.config.namesrvAddr=192.168.25.135:9876;192.168.25.138:9876

启动rocketmq-console：

java -jar rocketmq-console-ng-1.0.0.jar

启动成功后，我们就可以通过浏览器访问http://localhost:8080进入控制台界面了，如下图：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-o0hzVMv3-1681918953102)(assets/rocketmq-console2.png)]

集群状态：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-aAxryrDc-1681918953103)(assets/rocketmq-console3.png)]

2. 高级功能2.1 消息存储

分布式队列因为有高可靠性的要求，所以数据要进行持久化存储。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-gbI7fIkj-1681918953104)(assets/消息存储方式.png)]

消息生成者发送消息MQ收到消息，将消息进行持久化，在存储中新增一条记录返回ACK给生产者MQ push 消息给对应的消费者，然后等待消费者返回ACK如果消息消费者在指定时间内成功返回ack，那么MQ认为消息消费成功，在存储中删除消息，即执行第6步；如果MQ在指定时间内没有收到ACK，则认为消息消费失败，会尝试重新push消息,重复执行4、5、6步骤MQ删除消息2.1.1 存储介质关系型数据库DB

Apache下开源的另外一款MQ—ActiveMQ（默认采用的KahaDB做消息存储）可选用JDBC的方式来做消息持久化，通过简单的xml配置信息即可实现JDBC消息存储。由于，普通关系型数据库（如Mysql）在单表数据量达到千万级别的情况下，其IO读写性能往往会出现瓶颈。在可靠性方面，该种方案非常依赖DB，如果一旦DB出现故障，则MQ的消息就无法落盘存储会导致线上故障

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-zd44xhmC-1681918953106)(assets/MySQL.png)]

文件系统

目前业界较为常用的几款产品（RocketMQ/Kafka/RabbitMQ）均采用的是消息刷盘至所部署虚拟机/物理机的文件系统来做持久化（刷盘一般可以分为异步刷盘和同步刷盘两种模式）。消息刷盘为消息存储提供了一种高效率、高可靠性和高性能的数据持久化方式。除非部署MQ机器本身或是本地磁盘挂了，否则一般是不会出现无法持久化的故障问题。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-nzgEfRJV-1681918953107)(assets/磁盘.png)]

###2.1.2 性能对比

文件系统>关系型数据库DB

2.1.3 消息的存储和发送1）消息存储

磁盘如果使用得当，磁盘的速度完全可以匹配上网络 的数据传输速度。目前的高性能磁盘，顺序写速度可以达到600MB/s， 超过了一般网卡的传输速度。但是磁盘随机写的速度只有大概100KB/s，和顺序写的性能相差6000倍！因为有如此巨大的速度差别，好的消息队列系统会比普通的消息队列系统速度快多个数量级。RocketMQ的消息用顺序写,保证了消息存储的速度。

####2）消息发送

Linux操作系统分为【用户态】和【内核态】，文件操作、网络操作需要涉及这两种形态的切换，免不了进行数据复制。

一台服务器 把本机磁盘文件的内容发送到客户端，一般分为两个步骤：

1）read；读取本地文件内容；

2）write；将读取的内容通过网络发送出去。

这两个看似简单的操作，实际进行了4 次数据复制，分别是：

从磁盘复制数据到内核态内存；从内核态内存复 制到用户态内存；然后从用户态 内存复制到网络驱动的内核态内存；最后是从网络驱动的内核态内存复 制到网卡中进行传输。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-qVoJgyaZ-1681918953108)(assets/文件操作和网络操作.png)]通过使用mmap的方式，可以省去向用户态的内存复制，提高速度。这种机制在Java中是通过MappedByteBuffer实现的

RocketMQ充分利用了上述特性，也就是所谓的“零拷贝”技术，提高消息存盘和网络发送的速度。

这里需要注意的是，采用MappedByteBuffer这种内存映射的方式有几个限制，其中之一是一次只能映射1.5~2G 的文件至用户态的虚拟内存，这也是为何RocketMQ默认设置单个CommitLog日志数据文件为1G的原因了

2.1.4 消息存储结构

RocketMQ消息的存储是由ConsumeQueue和CommitLog配合完成 的，消息真正的物理存储文件是CommitLog，ConsumeQueue是消息的逻辑队列，类似数据库的索引文件，存储的是指向物理存储的地址。每 个Topic下的每个Message Queue都有一个对应的ConsumeQueue文件。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-3YlMHX5T-1681918953109)(assets/消息存储结构.png)]

CommitLog：存储消息的元数据ConsumerQueue：存储消息在CommitLog的索引IndexFile：为了消息查询提供了一种通过key或时间区间来查询消息的方法，这种通过IndexFile来查找消息的方法不影响发送与消费消息的主流程1.1.5 刷盘机制

RocketMQ的消息是存储到磁盘上的，这样既能保证断电后恢复， 又可以让存储的消息量超出内存的限制。RocketMQ为了提高性能，会尽可能地保证磁盘的顺序写。消息在通过Producer写入RocketMQ的时 候，有两种写磁盘方式，分布式同步刷盘和异步刷盘。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-MBmKNDYj-1681918953110)(assets/同步刷盘和异步刷盘.png)]

1）同步刷盘

在返回写成功状态时，消息已经被写入磁盘。具体流程是，消息写入内存的PAGECACHE后，立刻通知刷盘线程刷盘， 然后等待刷盘完成，刷盘线程执行完成后唤醒等待的线程，返回消息写 成功的状态。

2）异步刷盘

在返回写成功状态时，消息可能只是被写入了内存的PAGECACHE，写操作的返回快，吞吐量大；当内存里的消息量积累到一定程度时，统一触发写磁盘动作，快速写入。

####3）配置

同步刷盘还是异步刷盘，都是通过Broker配置文件里的flushDiskType 参数设置的，这个参数被配置成SYNC_FLUSH、ASYNC_FLUSH中的 一个。

2.2 高可用性机制

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-HdyOtyfd-1681918953111)(assets/RocketMQ角色-1602400165793.jpg)]

RocketMQ分布式集群是通过Master和Slave的配合达到高可用性的。

Master和Slave的区别：在Broker的配置文件中，参数 brokerId的值为0表明这个Broker是Master，大于0表明这个Broker是 Slave，同时brokerRole参数也会说明这个Broker是Master还是Slave。

Master角色的Broker支持读和写，Slave角色的Broker仅支持读，也就是 Producer只能和Master角色的Broker连接写入消息；Consumer可以连接 Master角色的Broker，也可以连接Slave角色的Broker来读取消息。

2.2.1 消息消费高可用

在Consumer的配置文件中，并不需要设置是从Master读还是从Slave 读，当Master不可用或者繁忙的时候，Consumer会被自动切换到从Slave 读。有了自动切换Consumer这种机制，当一个Master角色的机器出现故障后，Consumer仍然可以从Slave读取消息，不影响Consumer程序。这就达到了消费端的高可用性。

2.2.2 消息发送高可用

在创建Topic的时候，把Topic的多个Message Queue创建在多个Broker组上（相同Broker名称，不同 brokerId的机器组成一个Broker组），这样当一个Broker组的Master不可 用后，其他组的Master仍然可用，Producer仍然可以发送消息。 RocketMQ目前还不支持把Slave自动转成Master，如果机器资源不足， 需要把Slave转成Master，则要手动停止Slave角色的Broker，更改配置文 件，用新的配置文件启动Broker。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-gBlAHJLX-1681918953113)(assets/消息发送高可用设计.jpg)]

2.2.3 消息主从复制

如果一个Broker组有Master和Slave，消息需要从Master复制到Slave 上，有同步和异步两种复制方式。

####1）同步复制

同步复制方式是等Master和Slave均写 成功后才反馈给客户端写成功状态；

在同步复制方式下，如果Master出故障， Slave上有全部的备份数据，容易恢复，但是同步复制会增大数据写入 延迟，降低系统吞吐量。

####2）异步复制

异步复制方式是只要Master写成功 即可反馈给客户端写成功状态。

在异步复制方式下，系统拥有较低的延迟和较高的吞吐量，但是如果Master出了故障，有些数据因为没有被写 入Slave，有可能会丢失；

####3）配置

同步复制和异步复制是通过Broker配置文件里的brokerRole参数进行设置的，这个参数可以被设置成ASYNC_MASTER、 SYNC_MASTER、SLAVE三个值中的一个。

####4）总结

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-apTU9gMH-1681918953114)(assets/复制刷盘.png)]

实际应用中要结合业务场景，合理设置刷盘方式和主从复制方式， 尤其是SYNC_FLUSH方式，由于频繁地触发磁盘写动作，会明显降低 性能。通常情况下，应该把Master和Save配置成ASYNC_FLUSH的刷盘 方式，主从之间配置成SYNC_MASTER的复制方式，这样即使有一台 机器出故障，仍然能保证数据不丢，是个不错的选择。

2.3 负载均衡1.3.1 Producer负载均衡

Producer端，每个实例在发消息的时候，默认会轮询所有的message queue发送，以达到让消息平均落在不同的queue上。而由于queue可以散落在不同的broker，所以消息就发送到不同的broker下，如下图：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-q2PAfUuV-1681918953116)(assets/producer负载均衡.png)]

图中箭头线条上的标号代表顺序，发布方会把第一条消息发送至 Queue 0，然后第二条消息发送至 Queue 1，以此类推。

1.3.2 Consumer负载均衡1）集群模式

在集群消费模式下，每条消息只需要投递到订阅这个topic的Consumer Group下的一个实例即可。RocketMQ采用主动拉取的方式拉取并消费消息，在拉取的时候需要明确指定拉取哪一条message queue。

而每当实例的数量有变更，都会触发一次所有实例的负载均衡，这时候会按照queue的数量和实例的数量平均分配queue给每个实例。

默认的分配算法是AllocateMessageQueueAveragely，如下图：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-by6Kdr4W-1681918953117)(assets/consumer负载均衡.png)]

还有另外一种平均的算法是AllocateMessageQueueAveragelyByCircle，也是平均分摊每一条queue，只是以环状轮流分queue的形式，如下图：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-CkRP1yVt-1681918953118)(assets/consumer负载均衡2.png)]

需要注意的是，集群模式下，queue都是只允许分配只一个实例，这是由于如果多个实例同时消费一个queue的消息，由于拉取哪些消息是consumer主动控制的，那样会导致同一个消息在不同的实例下被消费多次，所以算法上都是一个queue只分给一个consumer实例，一个consumer实例可以允许同时分到不同的queue。

通过增加consumer实例去分摊queue的消费，可以起到水平扩展的消费能力的作用。而有实例下线的时候，会重新触发负载均衡，这时候原来分配到的queue将分配到其他实例上继续消费。

但是如果consumer实例的数量比message queue的总数量还多的话，多出来的consumer实例将无法分到queue，也就无法消费到消息，也就无法起到分摊负载的作用了。所以需要控制让queue的总数量大于等于consumer的数量。

####2）广播模式

由于广播模式下要求一条消息需要投递到一个消费组下面所有的消费者实例，所以也就没有消息被分摊消费的说法。

在实现上，其中一个不同就是在consumer分配queue的时候，所有consumer都分到所有的queue。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-EGvtxepG-1681918953119)(assets/consumer负载均衡3.png)]

2.4 消息重试2.4.1 顺序消息的重试

对于顺序消息，当消费者消费消息失败后，消息队列 RocketMQ 会自动不断进行消息重试（每次间隔时间为 1 秒），这时，应用会出现消息消费被阻塞的情况。因此，在使用顺序消息时，务必保证应用能够及时监控并处理消费失败的情况，避免阻塞现象的发生。

2.4.2 无序消息的重试

对于无序消息（普通、定时、延时、事务消息），当消费者消费消息失败时，您可以通过设置返回状态达到消息重试的结果。

无序消息的重试只针对集群消费方式生效；广播方式不提供失败重试特性，即消费失败后，失败消息不再重试，继续消费新的消息。

1）重试次数

消息队列 RocketMQ 默认允许每条消息最多重试 16 次，每次重试的间隔时间如下：

第几次重试与上次重试的间隔时间第几次重试与上次重试的间隔时间110 秒97 分钟230 秒108 分钟31 分钟119 分钟42 分钟1210 分钟53 分钟1320 分钟64 分钟1430 分钟75 分钟151 小时86 分钟162 小时

如果消息重试 16 次后仍然失败，消息将不再投递。如果严格按照上述重试时间间隔计算，某条消息在一直消费失败的前提下，将会在接下来的 4 小时 46 分钟之内进行 16 次重试，超过这个时间范围消息将不再重试投递。

注意： 一条消息无论重试多少次，这些重试消息的 Message ID 不会改变。

2）配置方式

消费失败后，重试配置方式

集群消费方式下，消息消费失败后期望消息重试，需要在消息监听器接口的实现中明确进行配置（三种方式任选一种）：

返回 Action.ReconsumeLater （推荐）返回 Null抛出异常public class MessageListenerImpl implements MessageListener { @Override public Action consume(Message message, ConsumeContext context) { //处理消息 doConsumeMessage(message); //方式1：返回 Action.ReconsumeLater，消息将重试 return Action.ReconsumeLater; //方式2：返回 null，消息将重试 return null; //方式3：直接抛出异常， 消息将重试 throw new RuntimeException("Consumer Message exceotion"); }}

消费失败后，不重试配置方式

集群消费方式下，消息失败后期望消息不重试，需要捕获消费逻辑中可能抛出的异常，最终返回 Action.CommitMessage，此后这条消息将不会再重试。

public class MessageListenerImpl implements MessageListener { @Override public Action consume(Message message, ConsumeContext context) { try { doConsumeMessage(message); } catch (Throwable e) { //捕获消费逻辑中的所有异常，并返回 Action.CommitMessage; return Action.CommitMessage; } //消息处理正常，直接返回 Action.CommitMessage; return Action.CommitMessage; }}

自定义消息最大重试次数

消息队列 RocketMQ 允许 Consumer 启动的时候设置最大重试次数，重试时间间隔将按照如下策略：

最大重试次数小于等于 16 次，则重试时间间隔同上表描述。最大重试次数大于 16 次，超过 16 次的重试时间间隔均为每次 2 小时。Properties properties = new Properties();//配置对应 Group ID 的最大消息重试次数为 20 次properties.put(PropertyKeyConst.MaxReconsumeTimes,"20");Consumer consumer =ONSFactory.createConsumer(properties);

注意：

消息最大重试次数的设置对相同 Group ID 下的所有 Consumer 实例有效。如果只对相同 Group ID 下两个 Consumer 实例中的其中一个设置了 MaxReconsumeTimes，那么该配置对两个 Consumer 实例均生效。配置采用覆盖的方式生效，即最后启动的 Consumer 实例会覆盖之前的启动实例的配置

获取消息重试次数

消费者收到消息后，可按照如下方式获取消息的重试次数：

public class MessageListenerImpl implements MessageListener { @Override public Action consume(Message message, ConsumeContext context) { //获取消息的重试次数 System.out.println(message.getReconsumeTimes()); return Action.CommitMessage; }}2.5 死信队列

当一条消息初次消费失败，消息队列 RocketMQ 会自动进行消息重试；达到最大重试次数后，若消费依然失败，则表明消费者在正常情况下无法正确地消费该消息，此时，消息队列 RocketMQ 不会立刻将消息丢弃，而是将其发送到该消费者对应的特殊队列中。

在消息队列 RocketMQ 中，这种正常情况下无法被消费的消息称为死信消息（Dead-Letter Message），存储死信消息的特殊队列称为死信队列（Dead-Letter Queue）。

2.5.1 死信特性

死信消息具有以下特性

不会再被消费者正常消费。有效期与正常消息相同，均为 3 天，3 天后会被自动删除。因此，请在死信消息产生后的 3 天内及时处理。

死信队列具有以下特性：

一个死信队列对应一个 Group ID， 而不是对应单个消费者实例。如果一个 Group ID 未产生死信消息，消息队列 RocketMQ 不会为其创建相应的死信队列。一个死信队列包含了对应 Group ID 产生的所有死信消息，不论该消息属于哪个 Topic。2.5.2 查看死信信息在控制台查询出现死信队列的主题信息

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-4rBFjh10-1681918953120)(assets/死信队列主题.png)]

在消息界面根据主题查询死信消息

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-SXgv8ayu-1681918953121)(assets/死信队列主题2.png)]

选择重新发送消息

一条消息进入死信队列，意味着某些因素导致消费者无法正常消费该消息，因此，通常需要您对其进行特殊处理。排查可疑因素并解决问题后，可以在消息队列 RocketMQ 控制台重新发送该消息，让消费者重新消费一次。

2.6 消费幂等

消息队列 RocketMQ 消费者在接收到消息以后，有必要根据业务上的唯一 Key 对消息做幂等处理的必要性。

2.6.1 消费幂等的必要性

在互联网应用中，尤其在网络不稳定的情况下，消息队列 RocketMQ 的消息有可能会出现重复，这个重复简单可以概括为以下情况：

发送时消息重复

当一条消息已被成功发送到服务端并完成持久化，此时出现了网络闪断或者客户端宕机，导致服务端对客户端应答失败。 如果此时生产者意识到消息发送失败并尝试再次发送消息，消费者后续会收到两条内容相同并且 Message ID 也相同的消息。

投递时消息重复

消息消费的场景下，消息已投递到消费者并完成业务处理，当客户端给服务端反馈应答的时候网络闪断。 为了保证消息至少被消费一次，消息队列 RocketMQ 的服务端将在网络恢复后再次尝试投递之前已被处理过的消息，消费者后续会收到两条内容相同并且 Message ID 也相同的消息。

负载均衡时消息重复（包括但不限于网络抖动、Broker 重启以及订阅方应用重启）

当消息队列 RocketMQ 的 Broker 或客户端重启、扩容或缩容时，会触发 Rebalance，此时消费者可能会收到重复消息。

2.6.2 处理方式

因为 Message ID 有可能出现冲突（重复）的情况，所以真正安全的幂等处理，不建议以 Message ID 作为处理依据。 最好的方式是以业务唯一标识作为幂等处理的关键依据，而业务的唯一标识可以通过消息 Key 进行设置：

Message message = new Message();message.setKey("ORDERID_100");SendResult sendResult = producer.send(message);

订阅方收到消息时可以根据消息的 Key 进行幂等处理：

consumer.subscribe("ons_test", "*", new MessageListener() { public Action consume(Message message, ConsumeContext context) { String key = message.getKey() // 根据业务唯一标识的 key 做幂等处理 }});

umer consumer =ONSFactory.createConsumer(properties);

> 注意：- 消息最大重试次数的设置对相同 Group ID 下的所有 Consumer 实例有效。- 如果只对相同 Group ID 下两个 Consumer 实例中的其中一个设置了 MaxReconsumeTimes，那么该配置对两个 Consumer 实例均生效。- 配置采用覆盖的方式生效，即最后启动的 Consumer 实例会覆盖之前的启动实例的配置**获取消息重试次数**消费者收到消息后，可按照如下方式获取消息的重试次数：```javapublic class MessageListenerImpl implements MessageListener { @Override public Action consume(Message message, ConsumeContext context) { //获取消息的重试次数 System.out.println(message.getReconsumeTimes()); return Action.CommitMessage; }}2.5 死信队列

当一条消息初次消费失败，消息队列 RocketMQ 会自动进行消息重试；达到最大重试次数后，若消费依然失败，则表明消费者在正常情况下无法正确地消费该消息，此时，消息队列 RocketMQ 不会立刻将消息丢弃，而是将其发送到该消费者对应的特殊队列中。

在消息队列 RocketMQ 中，这种正常情况下无法被消费的消息称为死信消息（Dead-Letter Message），存储死信消息的特殊队列称为死信队列（Dead-Letter Queue）。

2.5.1 死信特性

死信消息具有以下特性

不会再被消费者正常消费。有效期与正常消息相同，均为 3 天，3 天后会被自动删除。因此，请在死信消息产生后的 3 天内及时处理。

死信队列具有以下特性：

一个死信队列对应一个 Group ID， 而不是对应单个消费者实例。如果一个 Group ID 未产生死信消息，消息队列 RocketMQ 不会为其创建相应的死信队列。一个死信队列包含了对应 Group ID 产生的所有死信消息，不论该消息属于哪个 Topic。2.5.2 查看死信信息在控制台查询出现死信队列的主题信息

[外链图片转存中…(img-4rBFjh10-1681918953120)]

在消息界面根据主题查询死信消息

[外链图片转存中…(img-SXgv8ayu-1681918953121)]

选择重新发送消息

一条消息进入死信队列，意味着某些因素导致消费者无法正常消费该消息，因此，通常需要您对其进行特殊处理。排查可疑因素并解决问题后，可以在消息队列 RocketMQ 控制台重新发送该消息，让消费者重新消费一次。

2.6 消费幂等

消息队列 RocketMQ 消费者在接收到消息以后，有必要根据业务上的唯一 Key 对消息做幂等处理的必要性。

2.6.1 消费幂等的必要性

在互联网应用中，尤其在网络不稳定的情况下，消息队列 RocketMQ 的消息有可能会出现重复，这个重复简单可以概括为以下情况：

发送时消息重复

当一条消息已被成功发送到服务端并完成持久化，此时出现了网络闪断或者客户端宕机，导致服务端对客户端应答失败。 如果此时生产者意识到消息发送失败并尝试再次发送消息，消费者后续会收到两条内容相同并且 Message ID 也相同的消息。

投递时消息重复

消息消费的场景下，消息已投递到消费者并完成业务处理，当客户端给服务端反馈应答的时候网络闪断。 为了保证消息至少被消费一次，消息队列 RocketMQ 的服务端将在网络恢复后再次尝试投递之前已被处理过的消息，消费者后续会收到两条内容相同并且 Message ID 也相同的消息。

负载均衡时消息重复（包括但不限于网络抖动、Broker 重启以及订阅方应用重启）

当消息队列 RocketMQ 的 Broker 或客户端重启、扩容或缩容时，会触发 Rebalance，此时消费者可能会收到重复消息。

2.6.2 处理方式

因为 Message ID 有可能出现冲突（重复）的情况，所以真正安全的幂等处理，不建议以 Message ID 作为处理依据。 最好的方式是以业务唯一标识作为幂等处理的关键依据，而业务的唯一标识可以通过消息 Key 进行设置：

Message message = new Message();message.setKey("ORDERID_100");SendResult sendResult = producer.send(message);

订阅方收到消息时可以根据消息的 Key 进行幂等处理：

consumer.subscribe("ons_test", "*", new MessageListener() { public Action consume(Message message, ConsumeContext context) { String key = message.getKey() // 根据业务唯一标识的 key 做幂等处理 }});