rocketmq支持最大消息_RocketMQ核心概念扫盲篇

女爷i 2023-01-04 13:12 174阅读 0赞

点击上方“中间件兴趣圈”，选择“设为星标”

做积极的人，越努力越幸运！![56d66784f1c02d5ea4a62c5ff6afbcf5.png][]

在正式进入RocketMQ的学习之前，我觉得有必要梳理一下RocketMQ核心概念，为大家学习RocketMQ打下牢固的基础。

## **1、RocketMQ部署架构** ##

--------------------

![ba05779faa3a356c011b7bd332cd7fdb.png][]在RocketMQ主要的组件如下：

*  Nameserver
    
    Nameserver集群，topic的路由注册中心，为客户端根据Topic提供路由服务，从而引导客户端向Broker发送消息。Nameserver之间的节点不通信。路由信息在Nameserver集群中数据一致性采取的最终一致性。
 *  Broker
    
    消息存储服务器，分为两种角色：Master与Slave，上图中呈现的就是2主2从的部署架构，在RocketMQ中，主服务承担读写操作，从服务器作为一个备份，当主服务器存在压力时，从服务器可以承担读服务(消息消费)。所有Broker，包含Slave服务器每隔30s会向Nameserver发送心跳包，心跳包中会包含存在在Broker上所有的topic的路由信息。
 *  Client
    
    消息客户端，包括Producer(消息发送者)和Consumer(消费消费者)．客户端在同一时间只会连接一台nameserver，只有在连接出现异常时才会尝试连接另外一台。客户端每隔30s向Nameserver发起topic的路由信息查询。

> 温馨提示：Nameserver是在内存中存储Topic的路由信息，持久化Topic路由信息的地方是在Broker中，即$\{    ROCKETMQ\_HOME\}/store/config/topics.json。

在RocketMQ4.5.0版本后引入了多副本机制，即一个复制组(m-s)可以演变为基于raft协议的复制组，复制组内部使用raft协议保证broker节点数据的强一致性，该部署架构在金融行业用的比较多。

## **2、消息订阅模型** ##

--------------------

在RocketMQ的消息消费模式采用的是发布与订阅模式。

topic：一类消息的集合，消息发送者将一类消息发送到一个主题中，例如订单模块将订单发送到 order\_topic 中，而用户登录时，将登录事件发送到 user\_login\_topic 中。

consumegroup：消息消费组，一个消费单位的“群体”，消费组首先在启动时需要订阅需要消费的topic。一个topic可以被多个消费组订阅，同样一个消费组也可以订阅多个主题。一个消费组拥有多个消费者。

**术语解释起来有点枯燥晦涩，接下来我举例来阐述。**

例如我们在开发一个订单系统，其中有一个子系统：order-service-app，在该项目中会创建一个消费组order\_consumer来订阅 order\_topic，并且基于分布式部署，order-service-app的部署情况如下：

![701c52cbf8b611deb2684ecf47556664.png][]即order-service-app部署了3台服务器，每一个jvm进程可以看做是消费组 order\_consumer 消费组的其中一个消费者。

#### **2.1 消费模式** ####

那这三个消费者如何来分工来共同消费 order\_topic 中的消息呢？

在RocketMQ中支持广播模式与集群模式。

**广播模式**：一个消费组内的所有消费者每一个都会处理topic中的每一条消息，通常用于刷新内存缓存。

**集群模式**：一个消费组内的所有消费者共同消费一个topic中的消息，即分工协作，一个消费者消费一部分数据，启动负载均衡，

集群模式是非常普遍的模式，符合分布式架构的基本理念，即横向扩容，当前消费者如果无法快速及时处理消息时，可以通过增加消费者的个数横向扩容，快速提高消费能力，及时处理挤压的消息。

#### **2.2 消费队列负载算法与重平衡机制** ####

那集群模式下，消费者是如何来分配消息的呢？

例如上面实例中order\_topic有16个队列，那一个拥有3个消费者的消费组如何来分配队列中。

**在MQ领域有一个不成文的约定：同一个消费者同一时间可以分配多个队列，但一个队列同一时间只会分配给一个消费者。**

**RocketMQ提供了众多的队列负载算法**，其中最常用的两种平均分配算法。

*  AllocateMessageQueueAveragely
    
    平均分配
 *  AllocateMessageQueueAveragelyByCircle
    
    轮流平均分配

为了说明这两种分配算法的分配规则，现在对16个队列，进行编号，用q0~q15表示，消费者用c0～c2表示。

AllocateMessageQueueAveragely分配算法的队列负载机制如下：

c0：q0 q1 q2 q3 q4 q5

c1:   q6 q7 q8 q9 q10

c2:    q11 q12 q13 q14 q15

其算法的特点是用总数除以消费者个数，余数按消费者顺序分配给消费者，故c0会多分配一个队列，而且队列分配是连续的。

AllocateMessageQueueAveragelyByCircle分配算法的队列负载机制如下：

c0：q0  q3 q6 q9 q12 q15

c1:   q1   q4 q7 q10 q13

c2:    q2   q5 q8 q11 q14

该分配算法的特点就是轮流一个一个分配。

> 温馨提示：如果topic的队列个数小于消费者的个数，那有些消费者无法分配到消息。在RocketMQ中一个topic的队列数直接决定了最大消费者的个数，但topic队列个数的增加对RocketMQ的性能不会产生影响。

在实际过程中，对主题进行扩容(增加队列个数)或者对消费者进行扩容、缩容是一件非常寻常的事情，那如果新增一个消费者，该消费者消费哪些队列呢？这就涉及到消息消费队列的重新分配，即**消费队列重平衡机制**。

在RocketMQ客户端中会每隔20s去查询当前topic的所有队列、消费者的个数，运用队列负载算法进行重新分配，然后与上一次的分配结果进行对比，如果发生了变化，则进行队列重新分配；如果没有发生变化，则忽略。

例如采取的分配算法如下图所示，现在增加一个消费者c3，那队列的分布情况是怎样的呢？

![794302a454618add260a07d7acb076f2.png][]根据新的分配算法，其队列最终的情况如下：

c0：q0 q1 q2 q3

c1:   q4 q5 q6 q7

c2:    q8 q9 q10 q11

c3:    q12 q13 q14  q15

上述整个过程无需应用程序干预，由RocketMQ完成。大概的做法就是将将原先分配给自己但这次不属于的队列进行丢弃，新分配的队列则创建新的拉取任务。

#### **2.3 消费进度** ####

消费者消费一条消息后需要记录消费的位置，这样在消费端重启的时候，继续从上一次消费的位点开始进行处理新的消息。在RocketMQ中，消息消费位点的存储是以消费组为单位的。

**集群模式**下，消息消费进度存储在broker端，

不能识别此Latex公式:
     { ROCKETMQ_HOME }/store/config/consumerOffset.json 是其具体的存储文件，其中内容截图如下：
    
    可见消费进度的Key为：topic@consumeGroup，然后每一个队列一个偏移量。

`广播模式的消费进度文件存储在用户的主目录，默认文件全路劲名：`

`` \{ USER\_HOME \}/.rocketmq\_offsets。

#### **2.4 消费模型** ####

RocketMQ提供了并发消费、顺序消费两种消费模型。

**并发消费**：对一个队列中消息，每一个消费者内部都会创建一个线程池，对队列中的消息多线程处理，即偏移量大的消息比偏移量小的消息有可能先消费。

**顺序消费**：在某一项场景，例如MySQL binlog 场景，需要消息按顺序进行消费。在RocketMQ中提供了基于队列的顺序消费模型，即尽管一个消费组中的消费者会创建一个多线程，但针对同一个Queue，会加锁。

> 温馨提示：并发消费模型中，消息消费失败默认会重试16次，每一次的间隔时间不一样；而顺序消费，如果一条消息消费失败，则会一直消费，直到消费成功。故在顺序消费的使用过程中，应用程序需要区分系统异常、业务异常，如果是不符合业务规则导致的异常，则重试多少次都无法消费成功，这个时候一定要告警机制，及时进行人为干预，否则消费会积压。

## **3、事务消息** ##

--------------------

事务消息并不是为了解决分布式事务，而是提供消息发送与业务落库的一致性，其实现原理就是一次分布式事务的具体运用，请看如下示例：

![bd1721adb61ae56af20554bf958d8f39.png][]上述伪代码中，将订单存储关系型数据库中和将消息发送到MQ这是两个不同介质的两个操作，如果能保证消息发送、数据库存储这两个操作要么同时成功，要么同时失败，RocketMQ为了解决该问题引入了 **事务消息**。

> 温馨提示，本节主要的目的是让大家知晓各个术语的概念，由于事务消息的使用，将在该专栏的后续文章中详细介绍。

## **4、定时消息** ##

--------------------

开源版本的RocketMQ目前并不支持任意精度的定时消息。所谓的定时消息就是将消息发送到Broker，但消费端不会立即消费，而是要到指定延迟时间后才能被消费端消费。

RocketMQ目前支持指定级别的延迟，其延迟级别如下：

1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h

## **5、消息过滤** ##

--------------------

消息过滤是指消费端可以根据某些条件对一个topic中的消息进行过滤，即只消费一个主题下满足过滤条件的消息。

RocketMQ目前主要的过滤机制是基于tag的过滤与基于消息属性的过滤，基于消息属性的过滤支持SQL92表达式，对消息进行过滤。

## **6、小结** ##

--------------------

本文的主要目的是介绍RocketMQ常见的术语，例如nameserver、broker、主题、消费组、消费者、队列负载算法、队列重平衡机制、并发消费、顺序消费、消费进度存储、定时消息、事务消息、消息过滤等基本概念，为后续的实战系列打下坚实基础。

--------------------

本文来自笔者的另一力作《RocketMQ实战与进阶》，**专栏从使用场景入手介绍如何使用 RocketMQ，使用过程中遇到什么问题，如何解决这些问题，以及为什么可以这样解决，即原理讲解(图)穿插在实战中**。专栏的设计思路重在强调实战二字，旨在让一位 RocketMQ 初学者通过对本专栏的学习，快速“打怪升级”，理论与实战结合，成为该领域的佼佼者。

![5939f705a3f5f5098dd7bd0631637d43.png][]

[56d66784f1c02d5ea4a62c5ff6afbcf5.png]: /images/20221119/cfb690e2c71e4eddb8b89807a2abe17e.png
[ba05779faa3a356c011b7bd332cd7fdb.png]: https://img-blog.csdnimg.cn/img_convert/ba05779faa3a356c011b7bd332cd7fdb.png
[701c52cbf8b611deb2684ecf47556664.png]: https://img-blog.csdnimg.cn/img_convert/701c52cbf8b611deb2684ecf47556664.png
[794302a454618add260a07d7acb076f2.png]: https://img-blog.csdnimg.cn/img_convert/794302a454618add260a07d7acb076f2.png
[bd1721adb61ae56af20554bf958d8f39.png]: https://img-blog.csdnimg.cn/img_convert/bd1721adb61ae56af20554bf958d8f39.png
[5939f705a3f5f5098dd7bd0631637d43.png]: https://img-blog.csdnimg.cn/img_convert/5939f705a3f5f5098dd7bd0631637d43.png