当前位置：首页 > news >正文

消息中间件——RocketMQ（二）

news 来源：原创 2025/4/21 14:24:44

前言：此篇文章系本人学习过程中记录下来的笔记，里面难免会有不少欠缺的地方，诚心期待大家多多给予指教。

RocketMQ（一）

接上期内容：上期完成了RocketMQ单机部署知识。下面学习RocketMQ集群相关知识，话不多说，直接发车。

一、RocketMQ集群基础概念

（一）、核心组件

RocketMQ 集群主要由以下核心组件构成：

NameServer：RocketMQ 的名称服务器，为整个集群提供元数据管理服务。它保存了 Topic 与 Broker 的映射关系，以及 Broker 的状态信息等。NameServer 之间相互独立，没有通信和协作关系，采用无状态设计，方便水平扩展。
Broker：负责消息的存储、转发和查询等操作。Broker 与 NameServer 保持长连接，定时上报自身信息。根据功能不同，Broker 又可分为 Master 和 Slave 节点，Master 负责处理读写请求，Slave 主要用于数据备份和读请求分担。(主要集群的就是它)
Producer：消息生产者，负责将业务消息发送到 RocketMQ 集群。Producer 通过 NameServer 获取 Topic 对应的 Broker 列表，然后根据负载均衡策略将消息发送到具体的 Broker 节点。
Consumer：消息消费者，从 Broker 拉取或接收消息并进行业务处理。Consumer 支持集群消费和广播消费两种模式，集群消费模式下，同一个 Consumer Group 中的多个 Consumer 实例共同消费 Topic 的消息，提高消费效率；广播消费模式下，每个 Consumer 实例都会消费 Topic 的所有消息。

（二）、集群架构图

（三）、集群模式

RocketMQ 支持多种集群模式，常见的有：

单 Master 模式：整个集群只有一个 Master 节点，配置简单，但存在单点故障问题，一旦 Master 宕机，整个集群将无法工作，适用于本地测试或对可靠性要求不高的场景。
多 Master 模式：集群包含多个 Master 节点，每个 Master 节点之间相互独立，无 Slave 节点。这种模式性能较高，但每个 Master 节点宕机时，该节点上的消息无法被消费，适合对可用性要求不是特别高，追求高性能的场景。
多 Master 多 Slave 异步复制模式：每个 Master 节点都对应至少一个 Slave 节点，Master 与 Slave 之间采用异步复制方式同步数据。这种模式下，当 Master 宕机时，Slave 可以切换为 Master 继续提供服务，提高了集群的可用性，但可能存在数据丢失的风险。
多 Master 多 Slave 同步双写模式：与异步复制模式类似，但 Master 与 Slave 之间采用同步双写方式，即只有当 Master 和 Slave 都写入成功后，才认为消息写入成功。这种模式保证了数据的强一致性，即使 Master 宕机，也不会出现数据丢失，但性能会受到一定影响。

二、RocketMQ集群搭建

本次学习搭建一个双主双从异步复制的Broker集群。本应该是四台机器，由于硬件文件，所这里使用了两台主机来完成集群的搭建。

（一）、前期准备

准备好两天虚拟机，分别都装好RocketMQ、JDK。

（二）、修改配置

1、RocketMQ1

①、进入这个目录rocketmq-all-5.3.2-bin-release/conf/2m-2s-async

修改broker-a.properties文件

-a:代表Master，-s:代表salve。

# 指定整个broker集群的名称，或者说是RocketMQ集群的名称
brokerClusterName=DefaultCluster
# 指定master-slave集群的名称。一个RocketMQ集群可以包含多个master-slave集群
brokerName=broker-a
# master的brokerId为0
brokerId=0
# 指定删除消息存储过期文件的时间为凌晨4点
deleteWhen=04
# 指定未发生更新的消息存储文件的保留时长为48小时，48小时后过期，将会被删除
fileReservedTime=48
# 指定当前broker为异步复制master
brokerRole=ASYNC_MASTER
# 指定刷盘策略为异步刷盘
flushDiskType=ASYNC_FLUSH
# 指定Name Server的地址
namesrvAddr=MQ1IP:9876;MQ2IP:9876

复制策略：

复制策略是Broker的Master与Slave间的数据同步方式。分为同步复制与异步复制：

同步复制：消息写入master后，master会等待slave同步数据成功后才向producer返回成功ACK

异步复制：消息写入master后，master立即向producer返回成功ACK，无需等待slave同步数据成功。

异步复制策略会降低系统的写入延迟，RT变小，提高了系统的吞吐量。

刷盘策略：

刷盘策略指的是broker中消息的落盘(写入磁盘)方式，即消息发送到broker内存后消息持久化到磁盘的方式。分为同步刷盘与异步刷盘：

同步刷盘：当消息持久化到broker的磁盘后才算是消息写入成功。

异步刷盘：当消息写入到broker的内存后即表示消息写入成功，无需等待消息持久化到磁盘。

1）异步刷盘策略会降低系统的写入延迟，RT变小，提高了系统的吞吐量
2）消息写入到Broker的内存，一般是写入到了PageCache
3）对于异步刷盘策略，消息会写入到PageCache后立即返回成功ACK。但并不会立即做落盘操作，而是当PageCache到达一定量时会自动进行落盘。

②、修改broker-b-s.properties

# 指定整个broker集群的名称，或者说是RocketMQ集群的名称
brokerClusterName=DefaultCluster
# 指定这是另外一个master-slave集群
brokerName=broker-b
# slave的brokerId为非0
brokerId=1
deleteWhen=04
fileReservedTime=48
# 指定当前broker为slave
brokerRole=SLAVE
flushDiskType=ASYNC_FLUSH
namesrvAddr=MQ1IP:9876;MQ2IP:9876
# 指定Broker对外提供服务的端口，即Broker与producer与consumer通信的端口。默认
# 10911。由于当前主机同时充当着master1与slave2，而前面的master1使用的是默认端口。这
# 里需要将这两个端口加以区分，以区分出master1与slave2
listenPort=10912
# 指定消息存储相关的路径。默认路径为~/store目录。由于当前主机同时充当着master1与
# slave2，master1使用的是默认路径，这里就需要再指定一个不同路径
storePathRootDir=~/store-s
storePathCommitLog=~/store-s/commitlog
storePathConsumeQueue=~/store-s/consumequeue
storePathIndex=~/store-s/index
storeCheckpoint=~/store-s/checkpoint
abortFile=~/store-s/abort

2、RocketMQ2

修改内容跟RocketMQ1差距不大。

①、进入这个目录rocketmq-all-5.3.2-bin-release/conf/2m-2s-async

修改broker-b.properties文件

②、修改broker-a-s.properties

（三）、启动服务

1、先启动NameServer集群，命令都一样。

nohup sh bin/mqnamesrv &

2、在启动RocketMQ1与RocketMQ2两个主机中的broker master。注意，它们指定所要加载的配置文件是不同的。

# RocketMQ1
$ nohup sh bin/mqbroker -n xxx.xxx.xxx.xxx:9876 -c $ROCKETMQ_HOME/conf/2m-2s-async/broker-a.properties &
# RocketMQ2 
$ nohup sh bin/mqbroker -n xxx.xxx.xxx:9876 -c $ROCKETMQ_HOME/conf/2m-2s-async/broker-a.properties &

3、在启动RocketMQ1与RocketMQ2两个主机中的broker slave。注意，它们指定所要加载的配置文件是不同的。

# RocketMQ1
$ nohup sh bin/mqbroker -n xxx.xxx.xxx:9876 -c $ROCKETMQ_HOME/conf/2m-2s-async/broker-a-s.properties &
# RocketMQ2
$ nohup sh bin/mqbroker -n xxx.xxx.xxx:9876 -c $ROCKETMQ_HOME/conf/2m-2s-async/broker-a-s.properties &

4、RocketMQ控制台验证