RocketMQ进阶知识

迈不过友情╰ 2023-03-13 11:40 5阅读 0赞

### 文章目录 ###

*  一、消息存储
 *   *  1.1 储存介质
     *  1.2 消息的存储和发送
     *  1.3 消息存储结构
     *  1.4 刷盘机制
 *  二、高可用性机制
 *   *  2.1 RocketMQ 高可用性机制概述
     *  2.2 消息消费高可用
     *  2.3 消息发送高可用
     *  2.4 消息主从复制
 *  三、负载均衡
 *   *  3.1 Producer 负载均衡
     *  3.2 Consumer 负载均衡
 *  四、消息重试
 *   *  4.1 顺序消息的重试
     *  4.2 无序消息的重试
     *  4.3 重试次数
 *  五、死信队列
 *   *  5.1 死信队列概述
     *  5.2 查看死信信息
 *  六、消费幂等

# 一、消息存储 #

`MQ`通过集群化部署可以实现高可靠，如何消息丢失那么该怎么做?

需要对消息做持久化存储。  
![\[RocketMQ持久存储过程(img-ZGKF6oe1-1589336721459)(D:\\MyNote\\img.png)\]][RocketMQ_img-ZGKF6oe1-1589336721459_D_MyNote_img.png]

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

## 1.1 储存介质 ##

**①关系型数据库**

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

**②文件系统**

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

性能对比：文件系统>关系型数据库`DB`

## 1.2 消息的存储和发送 ##

**① 消息存储**

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

**②消息发送**

`Linux`操作系统分为【用户态】和【内核态】，文件操作、网络操作需要涉及这两种形态的切换，免不了进行数据复制。一台服务器把本机磁盘文件的内容发送到客户端，一般分为两个步骤：

1.  `read`：读取本地文件内容
2.  `write`：将读取的内容通过网络发送出去

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

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

![在这里插入图片描述][20200513101702374.png]  
通过使用`mmap`的方式，可以省去向用户态的内存复制，提高速度。这种机制在`Java`中是通过`MappedByteBuffer`实现的。`RocketMQ`充分利用了上述特性，也就是所谓的**零拷贝**技术，提高消息存盘和网络发的速度。

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

## 1.3 消息存储结构 ##

`RocketMQ`消息的存储是由`ConsumeQueue`和`CommitLog`配合完成的，消息真正的物理存储文件是`CommitLog`，`ConsumeQueue`是消息的逻辑队列，类似数据库的索引文件，存储的是指向物理存储的地址。每个`Topic`下的每个`Message Queue`都有一个对应的`ConsumeQueue`文件。  
![在这里插入图片描述][20200513104430797.png]

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

## 1.4 刷盘机制 ##

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

![在这里插入图片描述][20201015151109318.png]

**① 同步刷盘**

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

**②异步刷盘**

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

**③配置**

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

# 二、高可用性机制 #

## 2.1 RocketMQ 高可用性机制概述 ##

![在这里插入图片描述][20201015151505714.png]

`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 消息消费高可用 ##

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

## 2.3 消息发送高可用 ##

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

![在这里插入图片描述][2020051311361889.png]

## 2.4 消息主从复制 ##

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

**①同步复制**

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

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

**②异步复制**

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

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

**③配置**

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

**④总结**  
![在这里插入图片描述][20201015151722817.png]

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

# 三、负载均衡 #

## 3.1 Producer 负载均衡 ##

`Producer`端，每个实例在发消息的时候，默认会轮询所有的`message queue`发送，以达到让消息平均落在不同的`queue`上。而由于`queue`可以散落在不同的`broker`，所以消息就发送到不同的`broker`下，如下图：  
![在这里插入图片描述][20200513155509484.png]  
图中箭头线条上的标号代表顺序，发布方会把第一条消息发送至`Queue 0`，然后第二条消息发送至`Queue 1`，以此类推。

## 3.2 Consumer 负载均衡 ##

**① 集群模式**

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

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

默认的分配算法是`AllocateMessageQueueAveragely`，**每个`consumer`实例平均分配每个`consume queue`**，如下图：  
![在这里插入图片描述][20200513162134598.png]  
还有另外一种平均的算法是`AllocateMessageQueueAveragelyByCircle`，也是平均分摊每一条`queue`，只是以环状轮流分`queue`的形式，每个`consumer`实例平均分配每个`consume queue`，如下图：  
![在这里插入图片描述][20200513162420197.png]  
需要注意的是，集群模式下，`queue`都是只允许分配只一个实例，这是由于如果多个实例同时消费一个`queue`的消息，由于拉取哪些消息是`consumer`主动控制的，那样会导致同一个消息在不同的实例下被消费多次，所以算法上都是一个`queue`只分给一个`consumer`实例，一个`consumer`实例可以允许同时分到不同的`queue`。

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

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

**②广播模式**

由于广播模式下要求一条消息需要投递到一个消费组下面所有的消费者实例，所以也就没有消息被分摊消费的说法。在实现上，其中一个不同就是在`consumer`分配`queue`的时候，所有`consumer`都分到所有的`queue`，**每个`consumer`实例分配每个`consume queue`**。  
![在这里插入图片描述][20200513162900562.png]

# 四、消息重试 #

## 4.1 顺序消息的重试 ##

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

## 4.2 无序消息的重试 ##

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

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

## 4.3 重试次数 ##

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

<table> 
 <thead> 
  <tr> 
   <th>第几次重试</th> 
   <th>与上次重试的间隔时间</th> 
   <th>第几次重试</th> 
   <th>与上次重试的间隔时间</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>1</td> 
   <td>10 秒</td> 
   <td>9</td> 
   <td>7 分钟</td> 
  </tr> 
  <tr> 
   <td>2</td> 
   <td>30 秒</td> 
   <td>10</td> 
   <td>8分钟</td> 
  </tr> 
  <tr> 
   <td>3</td> 
   <td>1 分钟</td> 
   <td>11</td> 
   <td>9分钟</td> 
  </tr> 
  <tr> 
   <td>4</td> 
   <td>2 分钟</td> 
   <td>12</td> 
   <td>10分钟</td> 
  </tr> 
  <tr> 
   <td>5</td> 
   <td>3 分钟</td> 
   <td>13</td> 
   <td>20 分钟</td> 
  </tr> 
  <tr> 
   <td>6</td> 
   <td>4 分钟</td> 
   <td>14</td> 
   <td>30 分钟</td> 
  </tr> 
  <tr> 
   <td>7</td> 
   <td>5分钟</td> 
   <td>15</td> 
   <td>1 小时</td> 
  </tr> 
  <tr> 
   <td>8</td> 
   <td>6分钟</td> 
   <td>16</td> 
   <td>2 小时</td> 
  </tr> 
 </tbody> 
</table>

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

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

# 五、死信队列 #

## 5.1 死信队列概述 ##

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

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

**特点：**

*  不会再被消费者正常消费。
 *  有效期与正常消息相同，均为3天，3天后会被自动删除。因此，请在死信消息产生后的 3 天内及时处理。
 *  一个死信队列对应一个`Group ID`， 而不是对应单个消费者实例
 *  如果一个`Group ID`未产生死信消息，消息队列`RocketMQ`不会为其创建相应的死信队列。
 *  一个死信队列包含了对应`Group ID`产生的所有死信消息，不论该消息属于哪个`Topic`。

## 5.2 查看死信信息 ##

1.  在控制台查询出现死信队列的主题信息  
    ![在这里插入图片描述][20200513173212865.png]
2.  在消息界面根据主题查询死信消息  
    ![在这里插入图片描述][20200513173449830.png]
3.  选择重新发送消息  
    一条消息进入死信队列，意味着某些因素导致消费者无法正常消费该消息，因此，通常需要您对其进行特殊处理。排查可疑因素并解决问题后，可以在消息队列`RocketMQ`控制台重新发送该消息，让消费者重新消费一次。

# 六、消费幂等 #

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

*  **发送时消息重复**  
    当一条消息已被成功发送到服务端并完成持久化，此时出现了网络闪断或者客户端宕机，导致服务端对客户端应答失败。 如果此时生产者意识到消息发送失败并尝试再次发送消息，消费者后续会收到两条内容相同并且 `Message ID` 也相同的消息。
 *  **投递时消息重复**  
    消息消费的场景下，消息已投递到消费者并完成业务处理，当客户端给服务端反馈应答的时候网络闪断。为了保证消息至少被消费一次，消息队列`RocketMQ`的服务端 将在网络恢复后再次尝试投递之前已被处理过的消息，消费者后续会收到两条内容相同并且`Message ID`也相同的消息。
 *  **负载均衡时消息重复**（包括但不限于网络抖动、`Broker`重启以及订阅方应用重启）  
    当消息队列`RocketMQ`的`Broker`或客户端重启、扩容或缩容时，会触发`Rebalance`，此时消费者可能会收到重复消息。

**【解决方案】**

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

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

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

consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> { 
        try { 
            for (MessageExt msg : msgs) { 
                String key = msg.getKeys();
                StringUtils.isNotEmpty(redisTemplate.opsForValue().get(key)){ 
                    return new RuntimeException("含有重复性消费消息，请重试！");
                }
                redisTemplate.opsForValue().set(key, key);
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        } catch (Exception e) { 
            e.printStackTrace();
            //重试
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        } 
    });

[RocketMQ_img-ZGKF6oe1-1589336721459_D_MyNote_img.png]: /images/20230312/aabaaddcd1da4767acecfb8a1d5b6f3e.png
[20200513101702374.png]: /images/20230312/fc029b3792e84875803249682e1822a1.png
[20200513104430797.png]: https://img-blog.csdnimg.cn/20200513104430797.png
[20201015151109318.png]: https://img-blog.csdnimg.cn/20201015151109318.png
[20201015151505714.png]: https://img-blog.csdnimg.cn/20201015151505714.png
[2020051311361889.png]: https://img-blog.csdnimg.cn/2020051311361889.png
[20201015151722817.png]: https://img-blog.csdnimg.cn/20201015151722817.png
[20200513155509484.png]: https://img-blog.csdnimg.cn/20200513155509484.png
[20200513162134598.png]: https://img-blog.csdnimg.cn/20200513162134598.png
[20200513162420197.png]: https://img-blog.csdnimg.cn/20200513162420197.png
[20200513162900562.png]: https://img-blog.csdnimg.cn/20200513162900562.png
[20200513173212865.png]: https://img-blog.csdnimg.cn/20200513173212865.png
[20200513173449830.png]: https://img-blog.csdnimg.cn/20200513173449830.png