RocketMQ的原理 - 柳贯一的博客

引言

继续学习RokcetMQ。

RokcetMQ架构

RocketMQ开发官方文档：

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

RocketMQ的集群架构如下

RocketMQ架构上主要分为四部分，如上图所示

Producer

消息发布的角色，支持分布式集群方式部署。Producer通过nameserver的负载均衡模块选择相应的Broker集群队列进行消息投递，投递的过程支持快速失败并且低延迟。

Consumer

消息消费的角色，支持分布式集群方式部署。支持以push推，pull拉两种模式对消息进行消费。同时也支持集群方式和广播方式的消费，它提供实时消息订阅机制，可以满足大多数用户的需求。

Broker

Broker主要负责消息的存储、投递和查询以及服务高可用保证。

NameServer

NameServer是一个Broker与Topic路由的注册中心支持Broker的动态注册与发现主要包括两个功能。

Broker管理

NameServer接受Broker集群的注册信息并且保存下来作为路由信息的基本数据。然后提供心跳检测机制，检查Broker是否还存活。

路由信息管理

每个NameServer将保存关于Broker集群的整个路由信息和用于客户端查询的队列信息。然后Producer和Conumser通过NameServer就可以知道整个Broker集群的路由信息，从而进行消息的投递和消费。

RocketMQ入门

官方案例：https://github.com/apache/rocketmq/blob/master/docs/cn/RocketMQ_Example.md

导入依赖

注意和安装的MQ版本一致

<groupId>org.apache.rocketmq</groupId>

<artifactId>rocketmq-client</artifactId>

</dependency>

生产者

步骤分析

创建producer组

设置NameServer地址

start生产者

发送消息获取结果

结束producer

//消息发送者

public class ProducerTest {

public static void main(String[] args) {

try {

// 实例化消息生产者Producer

DefaultMQProducer producer = new DefaultMQProducer("producergroup");

// 设置NameServer的地址

producer.setNamesrvAddr("localhost:9876");

// 启动Producer实例

producer.start();

for (int i = 0; i < 100; i++) {

//构建消息

Message message = new Message("topic_log","tags_error",("我是消息"+i).getBytes());

SendResult sendResult = producer.send(message);

System.out.printf("%s%n", sendResult);

}

// 如果不再发送消息，关闭Producer实例。

producer.shutdown();

}catch (Exception e){

e.printStackTrace();

}

代码解释：

DefaultMQProducer ： MQ生产者，可以指定组名 producerGroupName

producer.setNamesrvAddr ：指定Name Server地址，用作Brocker发现。注意IP和启动name server服务时指定的IP保持一致。

producer.start() ：启动生产者

new Message(“topic_log”,“tags_error”,(“我是消息”+i).getBytes()) ：消息，参数为：topic，tags,内容

producer.send(message) : 发送消息

SendResult ：发送结果，其中包含

sendStatus=SEND_OK ：发送状态

msgId ：producer 创建的消息ID

offsetMsgId ：Brocker创建的消息ID

messageQueue ：消息存储的队列

producer.shutdown():关闭生产者

消费者

创建consumer组

设置Name Server地址

设置消费位置，从最开始销毁

设置消息回调处理监听 -> 处理消息

Start consumer

//消息发送者

public class ConsumerTest {

public static void main(String[] args) {

try {

// 实例化消息生产者Producer

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer ("consumergroup");

// 设置NameServer的地址

consumer.setNamesrvAddr("127.0.0.1:9876");

//从最开始的位置开始消费

consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

// 订阅一个或者多个Topic，以及Tag来过滤需要消费的消息

//和发送者保持一致才能搜到消息

consumer.subscribe("topic_log", "tags_error");

// 注册回调实现类来处理从broker拉取回来的消息

consumer.registerMessageListener(new MessageListenerConcurrently() {

@Override

public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,

ConsumeConcurrentlyContext context) {

System.out.printf("%s 成功搜到消息: %s %n", Thread.currentThread().getName(), msgs);

// 标记该消息已经被成功消费

return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;

}

});

// 启动Producer实例

consumer.start();

}catch (Exception e){

e.printStackTrace();

}

DefaultMQPushConsumer ：消费者，可以指定 consumerGroupName

consumer.setNamesrvAddr : 设置name server 地址

consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET) ：从什么位置开始消费

consumer.subscribe(“topic_log”, “tags_error”) ：订阅某个topic下的某个tags的消息

consumer.registerMessageListener ：注册消息监听器，拿到消息后，进行消息处理。

ConsumeConcurrentlyStatus ：消费者消费结果状态，ConsumeConcurrentlyStatus.CONSUME_SUCCESS代表成功

RocketMQ 核心概念

RocketMQ工作原理

1.RocketMQ 网络部署特点

为了增强Broker性能与吞吐量，Broker一般都是以集群形式出现的。各集群节点中可能存放着相同Topic的不同Queue。

不过，这里有个问题，如果某Broker节点宕机，如何保证数据不丢失呢？其解决方案是，将每个Broker集群节点进行横向扩展，即将Broker节点再建为一个集群，解决单点问题。

Broker节点集群是一个主从集群，即集群中具有Master与Slave两种角色。Master负责处理读写操作请求，Slave负责对Master中的数据进行备份。当Master挂掉了，Slave则会自动切换为Master去工作。所以这个Broker集群是主备集群。Consumer既可以从Master订阅消息，也可以从Slave订阅消息。

一个Master可以包含多个Slave，但一个Slave只能隶属于一个Master。 Maste与Slave 的对应关系是通过指定相同的BrokerName、不同的BrokerId 来确定的。BrokerId为0表示Master非0表示Slave。每个Broker与NameServer集群中的所有节点建立长连接，定时注册Topic信息到所有NameServer。

RocketMQ工作流程

启动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发消息。

topicid broker 队列

topic_log 1 队列1,队列2

topic_log 2 队列3,队列4

Consumer跟Producer类似，跟其中一台NameServer建立长连接，获取当前订阅Topic存在哪些Broker上，然后直接跟Broker建立连接通道，开始消费消息。

Producer 生产者

RocketMQ提供多种发送方式，同步发送、异步发送、顺序发送、单向发送。同步和异步方式均需要Broker返回确认信息，单向发送不需要。

RocketMQ中的消息生产者都是以生产者组（Producer Group）的形式出现的。生产者组是同一类生产者的集合，这类Producer发送相同Topic类型的消息。一个生产者组可以同时发送多个主题的消息。

Producer会使用一定的算法(随机轮询+规避故障)选择把消息发送到哪个master的某个queue中。

Consumer 消费者

Consumer 支持两种消费形式：拉取式消费、推动式消费。（主动，被动），RocketMQ中的消息消费者都是以消费者组（Consumer Group）的形式出现的。消费者组是同一类消费者的集合，这类Consumer消费的是同一个Topic类型的消息，不同的 Consumer Group可以消费同一个Topic。

一个Consumer Group内的Consumer可以消费多个Topic的消息。

[注意] 集群模式:一个Queue是不能被同一个ConsumerGroup中的多个Consumer消费的，目的是减少资源竞争提升整体性能。

Topic 消息主题

Topic表示一类消息的集合，每个topic主题包含若干条message消息，每条message消息只能属于一个topic主题，Topic是RocketMQ进行消息订阅的基本单位。

Message

消息是指消息系统所传输信息的物理载体，生产和消费数据的最小单位，每条消息必须属于一个主题。

Tag 标签

为消息设置的标志，用于同一主题下区分不同类型的消息。来自同一业务单元的消息，可以根据不同业务目的在同一主题下设置不同标签。标签能够有效地保持代码的清晰度和连贯性，并优化RocketMQ提供的查询系统。消费者可以根据Tag实现对不同子主题的不同消费逻辑，实现更好的扩展性。Topic是消息的一级分类，Tag是消息的二级分类

MessageQueue

一个Topic中可以包含多个Queue，一个Topic的Queue也被称为一个Topic中消息的分区（Partition）。

在集群模式下, 在一个Consumer Group内，一个Queue最多只能分配给一个Consumer，一个Cosumer可以分配得到多个Queue。这样的分配规则，每个Queue只有一个消费者，可以避免消费过程中的多线程处理和资源锁定，有效提高各Consumer消费的并行度和处理效率。

消费者组中Consumer的数量应该小于等于订阅Topic的Queue数量。如果超出Queue数量，则多出的 Consumer将不能消费消息。如果一个Consmer挂了，该Consumer Group中的其它Consumer可以接着消费原Consumer消费的Queue。

【注意】一个Topic可以对应多个消费者，一个Queue只能对应一个组中的一个消费者。

【注意】为了防止消息紊乱，一个Consumer Group 中的Consumer都是订阅相同Topic下的Queue。

在广播模式下一个队列要可以被多个消费者对应.在广播模式下，同一个 ConsumerGroup 中的每个 Consumer 监听全部的队列。需要注意的是，广播模式下因为每个 Consumer 实例都需要处理全部的消息，因此这种模式仅推荐在**通知推送、配置同步类小流量场景使用。

读写队列

Queue分为写队列(对于生产者来说是写) 和读队列(对于消费者来说) ，默认创建数量是都是4 ，这个读写队列是从逻辑上进行划分在物理上读/写是一个队列，Producer发送的消息进入写队列，Consumer从读队列获取数据，一半情况下读写队列数量是一样的。

可以通过可视化界面修改Topic中的队列数量

perm用于设置对当前创建Topic的操作权限：2表示只写，4表示只读，6表示读写。

MessageId/Key

RocketMQ中每个消息拥有唯一的MessageId，且可以携带具有业务标识的Key，以方便对消息的查询。不过需要注意的是，MessageId有两个：在生产者send()消息时会自动生成一个MessageId（msgId)，

当消息到达Broker后，Broker也会自动生成一个MessageId(offsetMsgId)。msgId、offsetMsgId与key都称为消息标识。

Rebalance重新负载

当消费者数量或者Queue的数量修改，Rebalance是把⼀个Topic下的多个Queue重新分配给Consumer Group下的Consumer。目的是增加消费能力。

由于一个队列值分配给一个Consumer,那么当Consumer Group中的消费者数量大于队列数量，那么多出来的Consumer分配不到队列。

消息拉取模式

消息的消费分为：拉取式 pull ,和推送是 push

Pull:拉取式，需要消费者间隔一定时间就去遍历关联的Queue,实时性差但是便于应用控制消息的拉取。

Push:推送式，封装了Queue的遍历，实时性强，但是对系统资源占用比较多。

消息消费模式

广播模式(一对多):同一个Consumer Group 下的所有Consumer都会受到同一个Topic的所有消息。同一个消息可能会被消费多次。

集群模式(一对一):同一个Gonsumer Group 下的Consumer平分同一个Topic下的消息。同一个消息只是被消费一次。

Queue的分配算法

Queue是如何分配给Consumer的，这对应了四种算法：平均分配策略，环形平均策略，一致性Hash策略，同机房策略。

平均分配【默认】：根据 qeueuCount (4)/ consumerCount (3)作为每个消费者平均分配数量，如果多出来的queue就再依次逐个分配给Consumer。

环形平均策略：根据消费者的顺序，一个一个的分配Queue即可类似于发扑克牌。

一致性Hash策略：该算法将Consumer的Hash值作为节点放到Hash环上，然后将Queue的hash值也放入Hash环上，通过顺时针进行就近分配。

同机房策略：该算法会根据queue的部署机房位置和consumer的位置，过滤出当前consumer相同机房的queue。然后按照平均分配策略或环形平均策略对同机房queue进行分配。如果没有同机房queue，则按照平均分配策略或环形平均策略对所有queue进行分配。

平均分配性能比较高，一致性Hash性能不高，但是能减少Rebalance，如果Consumer数量变动频繁可以使用一致性Hash。

Offset管理

RockertMQ通过Offset来维护Consumer的消费进度，比如：消费者从哪个位置开始持续消费消息的？这里有三个枚举来指定从什么位置消费

CONSUME_FROM_LAST_OFFSET：从queue的最后一条消息开始消费

CONSUME_FROM_FIRST_OFFSET：从queue的第一条消息开始消费

CONSUME_FROM_TIMESTAMP：从某个时间戳位置的消息开始消费。

消费者消费结束之后，会向Consumer会提交其消费进度offset给Broker。Offset信息的存储分为本地 Offset管理和远程Offset管理

远程Offset管理:Brocker通过 store/config/consumerOffset.json 文件以JSON方式来存储offset相关数据以json的形式：适用于集群模式

本地Offset管理：offset相关数据以json的形式持久化到Consumer本地磁盘文件中，路径为当前用户主目录下的.rocketmq_offsets/ c l i e n t I d / {clientId}/ clientId/{group}/Offsets.json ：适用于广播模式

Offset的同步提交与异步提交: 集群消费模式下，Consumer消费完消息后会向Broker提交消费进度offset，其提交方式分为两种：

同步提交：消费者在消费完一批消息后会向broker提交这些消息的offset，等待broker的成功响应。若在等待超时之前收到了成功响应，则继续读取下一批消息进行消费（从ACK中获取 nextBeginOffset）。若没有收到响应，则会重新提交，直到获取到响应。而在这个等待过程中，消费者是阻塞的。其严重影响了消费者的吞吐量。

异步提交：消费者在消费完一批消息后向broker提交offset，但无需等待Broker的成功响应，可以继续读取并消费下一批消息。这种方式增加了消费者的吞吐量。但需要注意，broker在收到提交的offset 后，还是会向消费者进行响应的。可能还没有收到ACK，此时Consumer会从Broker中直接获取 nextBeginOffset。

消息的清理

消息不会被单独清理，消息是顺序存储到commitlog的，消息是以commitlog为单位进行清理，RocketMQ有自己的清理规则，默认是72小时后进行清理

到达时间清理点，自动清理过期的文件(凌晨4点)

磁盘空间使用率达到了过期清理阈值(75%),自动清理过期的文件。

磁盘占用率达到清理阈值(85%)，开始按照设定的规则清理文件，从老的文件开始。

磁盘占用率达到系统危险阈值(90%),拒绝写入数据。

最后

感谢原作者的精彩作品，我受益匪浅

原文链接：https://blog.csdn.net/weixin_48133130/article/details/134126430