RocketMQ集群详解

集群特点
(1)NameServer：是一个几乎无状态节点，可集群部署，节点之间无任何信息同步。
(2)Broker：部署相对复杂，Broker分为Master与Slave，一个Master可以对应多个Slave，但是一个Slave只能对应一个Master，Master与Slave的对应关系通过指定相同的BrokerName，不同的BrokerId来定义，BrokerId为0表示Master，非0表示Slave。Master也可以部署多个。每个Broker与NameServer集群中的所有节点建立长连接，定时注册Topic信息到所有NameServer。
(3)Producer：与NameServer集群中的其中一个节点（随机选择）建立长连接，定期从NameServer取Topic路由信息，并向提供Topic服务的Master建立长连接，且定时向Master发送心跳。Producer完全无状态，可集群部署。
(4)Consumer：与NameServer集群中的其中一个节点（随机选择）建立长连接，定期从NameServer取Topic路由信息，并向提供Topic服务的Master、Slave建立长连接，且定时向Master、Slave发送心跳。Consumer既可以从Master订阅消息，也可以从Slave订阅消息，订阅规则由Broker配置决定。

集群工作流程
(1)启动NameServer，NameServer起来后监听端口，等待Broker、Producer、Consumer连上来，相当于一个路由控制中心。
(2) Broker启动，跟所有的NameServer保持长连接，定时发送心跳包。心跳包中包含当前Broker信息(IP+端口等)以及存储所有Topic信息。注册成功后，NameServer集群中就有Topic跟Broker的映射关系。
(3)收发消息前，先创建Topic，创建Topic有2种方法
(3.1)手动创建Topic：NameServer服务器上rocketmq的bin目录通过指令新增或者consolse后台添加
(3.2)自动创建Topic：broker.properties配置autoCreateTopicEnable=true,Broker会预先创建【"BW102"topic】并将其注册在NameServer上，Producer发送消息时，
如果是第一次发送该Topic，那么在Nam-eServer上找不到该Topic的注册信息，此时会默认用【"BW102"topic】的路由信息,Topic名称还是Producer的名称，只是"盗用"了【"BW102"topic】的路由信息，
下一次Producer发送该Topic时，就不需要用默认的【"BW102"topic】的路由信息了。
(4)Producer发送消息，启动时先跟NameServer集群中的其中一台建立长连接，并从NameServer中获取当前发送的Topic存在哪些Broker上，轮询从队列列表中选择一个队列，然后与队列所在的Broker建立长连接从而向Broker发消息。
(5)Consumer跟Producer类似，跟其中一台NameServer建立长连接，获取当前订阅Topic存在哪些Broker上，然后直接跟Broker建立连接通道，开始消费消息。

关于同步异步
(1)同步复制：消息进入master节点之后，master会把数据复制到slave节点，复制成功之后再会给客户端返回成功消息，效率较低，但是数据安全性高（数据安全性高，性能低一点）
(2)异步复制：消息发送到master节点之后，master将给客户端返还成功消息，并异步的将消息复制到slave节点，效率较高，但是由于是异步复制，不能百分百保证数据复制成功（数据可能丢失，性能高）
(3)同步刷盘：消息进来之后，broker获取到消息，内存中消息一有更新，马上会往磁盘里头写入数据
(4)异步刷盘：异步刷盘数据可能丢失，但是性能会更高，因为同步刷盘是消息进入内存之后，不会马上返回，等把数据存入磁盘之后才会返回成功，异步刷盘是把消息存入内存之后就会马上返回，异步将数据存入磁盘中，所以异步刷盘性能会比同步刷盘性能更高

RocketMQ 集群部署模式
单 master 模式：
优点：除了配置简单没什么优点，适合个人学习使用。

缺点：不可靠，该机器重启或宕机，将导致整个服务不可用。

多 master 模式：
多个 master 节点组成集群，单个 master 节点宕机或者重启对应用没有影响。

优点：所有模式中性能最高

缺点：单个 master 节点宕机期间，未被消费的消息在节点恢复之前不可用，消息的实时性就受到影响。

注意：使用同步刷盘可以保证消息不丢失，同时 Topic 相对应的 queue 应该分布在集群中各个节点，而不是只在某各节点上，否则，该节点宕机会对订阅该 topic 的应用造成影响。

多 master 多 slave 异步复制模式：
在多 master 模式的基础上，每个 master 节点都有至少一个对应的 slave。master
节点可读可写，但是 slave 只能读不能写，类似于 mysql 的主备模式。

优点：在 master 宕机时，消费者可以从 slave读取消息，消息的实时性不会受影响，性能几乎和多 master 一样。

缺点：使用异步复制的同步方式有可能会有消息丢失的问题。

多 master 多 slave 同步双写模式：
同多 master 多 slave 异步复制模式类似，区别在于 master 和 slave 之间的数据同步方式。

优点：同步双写的同步模式能保证数据不丢失。

缺点：发送单个消息 RT 会略长，性能相比异步复制低10%左右。

刷盘策略：同步刷盘和异步刷盘（指的是节点自身数据是同步还是异步存储）

同步方式：同步双写和异步复制（指的一组 master 和 slave 之间数据的同步）

注意：要保证数据可靠，需采用同步刷盘和同步双写的方式，但性能会较其他方式低

Broker

Master如何同步到Slave

这里注意一下，所有的Slave也会向所有的NameServer进行注册，也会30s发送心跳。

RocketMQ自身的Master-Slave模式采取的是Pull模式拉取信息，也就是Slave Broker不停的发送请求到Master Broker去拉取信息

RocketMQ实现读写分离了吗
原则上不算是完全的读写分离，生产者发送消息是往Master进行发送，而消费者获取消息则有可能是Master有可能是Slave，要根据Master Broker的负载情况和Slave Broker的同步情况判断。

如果Master Broker负载很重，已经要抗10w写并发，那么此时Master Broker就会建议你从Slave Broker中拉取消息，还有就是Slave同步的比较慢，100w数据差了4w，而消费者可能都获取完96w了，那么下次还是只能从Master Broker去拉取消息，因为同步太慢，没法获取更新的消息了。

Master或者Slave挂掉了有什么影响
1.Master Broker宕机

这个时候Slave Broker有一样的数据在的，只不过可能会有部分数据没来得及同步过来，而且不能自动切换成Master Broker。

4.5版本前，一旦Master故障，那么就需要手动做一些运维操作，将Slave重新修改参数配置，重启机器调整为Master Broker，所以这种Master-Slave模式不是彻底的高可用模式，没法自动切换主从。

4.5版本后，引入了新的机制，叫做Dledger。

这个机制是基于Raft协议实现，简单来说把Dledger融入RocketMQ之后，可以让一个Master Broker对应多个Slave Broker，也就是一份数据有多份备份，一旦Master宕机了，那么就可以在多个Slave中，通过Dledger技术和Raft协议算法选举处leader，然后直接将一个Slave Broker选举成新的Master Broker，整个过程也许只要10s或者几十秒就可以自动完成。

2.Slave Broker宕机

会有一点影响，但是不大，因为写入全部发送到Master Broker中，然后获取也是可以走Master Broker的，所以整体影响不大，只不过会导致读写压力都集中在Master Broker上。

下面采用的就是【异步复制，异步刷盘】的模式：

1.服务器环境

IP	角色	架构模式
192.168.247.129	nameserver1	broker-master1
192.168.247.130	nameserver2	broker-slave1
192.168.247.131	nameserver3	broker-master2
192.168.247.132	nameserver4	broker-slave2

2.4台服务器都按照上面步骤安装rocketmq

3.开放4台服务器的rocketmq防火墙端口

firewall-cmd --permanent --add -port=9876/tcp
firewall-cmd --permanent --add -port=10909/tcp
firewall-cmd --permanent --add -port=10911/tcp
firewall-cmd --reload

4.修改4台服务器的2m-2s-async目录下面对应的配置文件，这里根据主从不同，替换的配置文件也不同

（1）broker-master1

服务器：192.168.247.129

配置文件：broker-a.properties

配置内容：

#所属集群名字
brokerClusterName=rocketmq-cluster
#broker名字，注意此处不同的配置文件填写的不一样
brokerName=broker-a
#0 表示 Master，>0 表示 Slave
brokerId=0
#nameServer地址，分号分割
namesrvAddr=192.168.247.129:9876;192.168.247.130:9876;192.168.247.131:9876;192.168.247.132:9876
#在发送消息时，自动创建服务器不存在的topic，默认创建的队列数
defaultTopicQueueNums=4
#是否允许 Broker 自动创建Topic，建议线下开启，线上关闭
autoCreateTopicEnable=true
#是否允许 Broker 自动创建订阅组，建议线下开启，线上关闭
autoCreateSubscriptionGroup=true
#Broker 对外服务的监听端口
listenPort=10911
#删除文件时间点，默认凌晨 4点
deleteWhen=04
#文件保留时间，默认 48 小时
fileReservedTime=120
#commitLog每个文件的大小默认1G
mapedFileSizeCommitLog=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
#- SLAVE
brokerRole=ASYNC_MASTER
#刷盘方式
#- ASYNC_FLUSH 异步刷盘
#- SYNC_FLUSH 同步刷盘
flushDiskType=ASYNC_FLUSH
#checkTransactionMessageEnable=false
#发消息线程池数量
#sendMessageThreadPoolNums=128
#拉消息线程池数量
#pullMessageThreadPoolNums=128

（2）broker-slave1

服务器：192.168.247.130

配置文件：broker-a-s.properties

配置内容：

#所属集群名字
brokerClusterName=rocketmq-cluster
#broker名字，注意此处不同的配置文件填写的不一样
brokerName=broker-a
#0 表示 Master，>0 表示 Slave
brokerId=1
#nameServer地址，分号分割
namesrvAddr=192.168.247.129:9876;192.168.247.130:9876;192.168.247.131:9876;192.168.247.132:9876
#在发送消息时，自动创建服务器不存在的topic，默认创建的队列数
defaultTopicQueueNums=4
#是否允许 Broker 自动创建Topic，建议线下开启，线上关闭
autoCreateTopicEnable=true
#是否允许 Broker 自动创建订阅组，建议线下开启，线上关闭
autoCreateSubscriptionGroup=true
#Broker 对外服务的监听端口
listenPort=10911
#删除文件时间点，默认凌晨 4点
deleteWhen=04
#文件保留时间，默认 48 小时
fileReservedTime=120
#commitLog每个文件的大小默认1G
mapedFileSizeCommitLog=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
#- SLAVE
brokerRole=SLAVE
#刷盘方式
#- ASYNC_FLUSH 异步刷盘
#- SYNC_FLUSH 同步刷盘
flushDiskType=ASYNC_FLUSH
#checkTransactionMessageEnable=false
#发消息线程池数量
#sendMessageThreadPoolNums=128
#拉消息线程池数量
#pullMessageThreadPoolNums=128

（3）broker-master2

服务器：192.168.247.131

配置文件：broker-b.properties

配置内容：

#所属集群名字
brokerClusterName=rocketmq-cluster
#broker名字，注意此处不同的配置文件填写的不一样
brokerName=broker-b
#0 表示 Master，>0 表示 Slave
brokerId=0
#nameServer地址，分号分割
namesrvAddr=192.168.247.129:9876;192.168.247.130:9876;192.168.247.131:9876;192.168.247.132:9876
#在发送消息时，自动创建服务器不存在的topic，默认创建的队列数
defaultTopicQueueNums=4
#是否允许 Broker 自动创建Topic，建议线下开启，线上关闭
autoCreateTopicEnable=true
#是否允许 Broker 自动创建订阅组，建议线下开启，线上关闭
autoCreateSubscriptionGroup=true
#Broker 对外服务的监听端口
listenPort=10911
#删除文件时间点，默认凌晨 4点
deleteWhen=04
#文件保留时间，默认 48 小时
fileReservedTime=120
#commitLog每个文件的大小默认1G
mapedFileSizeCommitLog=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
#- SLAVE
brokerRole=ASYNC_MASTER
#刷盘方式
#- ASYNC_FLUSH 异步刷盘
#- SYNC_FLUSH 同步刷盘
flushDiskType=ASYNC_FLUSH
#checkTransactionMessageEnable=false
#发消息线程池数量
#sendMessageThreadPoolNums=128
#拉消息线程池数量
#pullMessageThreadPoolNums=128

（4）broker-slave2

服务器：192.168.247.132

配置文件：broker-b-s.properties

配置内容：

#所属集群名字
brokerClusterName=rocketmq-cluster
#broker名字，注意此处不同的配置文件填写的不一样
brokerName=broker-b
#0 表示 Master，>0 表示 Slave
brokerId=1
#nameServer地址，分号分割
namesrvAddr=192.168.247.129:9876;192.168.247.130:9876;192.168.247.131:9876;192.168.247.132:9876
#在发送消息时，自动创建服务器不存在的topic，默认创建的队列数
defaultTopicQueueNums=4
#是否允许 Broker 自动创建Topic，建议线下开启，线上关闭
autoCreateTopicEnable=true
#是否允许 Broker 自动创建订阅组，建议线下开启，线上关闭
autoCreateSubscriptionGroup=true
#Broker 对外服务的监听端口
listenPort=10911
#删除文件时间点，默认凌晨 4点
deleteWhen=04
#文件保留时间，默认 48 小时
fileReservedTime=120
#commitLog每个文件的大小默认1G
mapedFileSizeCommitLog=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
#- SLAVE
brokerRole=SLAVE
#刷盘方式
#- ASYNC_FLUSH 异步刷盘
#- SYNC_FLUSH 同步刷盘
flushDiskType=ASYNC_FLUSH
#checkTransactionMessageEnable=false
#发消息线程池数量
#sendMessageThreadPoolNums=128
#拉消息线程池数量
#pullMessageThreadPoolNums=128

5.启动服务

先启动nameserver:4台服务器都相同

cd /usr/local/rocketmq/bin/
nohup sh mqnamesrv >/dev/null 2>&1 &

再启动broker:4台服务器分别启动

cd /usr/local/rocketmq/bin/
nohup sh mqbroker -c /usr/local/rocketmq/conf/2m-2s-async/broker-a.properties > /dev/null 2>&1 &
nohup sh mqbroker -c /usr/local/rocketmq/conf/2m-2s-async/broker-a-s.properties > /dev/null 2>&1 &
nohup sh mqbroker -c /usr/local/rocketmq/conf/2m-2s-async/broker-b.properties > /dev/null 2>&1 &
nohup sh mqbroker -c /usr/local/rocketmq/conf/2m-2s-async/broker-b-s.properties > /dev/null 2>&1 &

打开管理后台界面能看到4台服务器集群

/*【安装RocketMq-Externals监控平台】/

1.github下载监控平台源码：

https://github.com/apache/rocketmq-externals

2.解压后打开文件夹

3.修改application.properties配置文件的namesrvAddr为实际的namesrv地址

4.修改配置后需要重新编译jar包，切换到target目录，并执行rocketmq-console-ng-1.0.1.jar包，浏览器输入http://localhost:8080 就能看到管理页面，默认端口也可以改

注意：如果是本机控制台访问服务器的rocketmq，需要开放服务器的几个端口

firewall-cmd --permanent --add -port=9876/tcp
firewall-cmd --permanent --add -port=10909/tcp
firewall-cmd --permanent --add -port=10911/tcp
firewall-cmd --reload