RocketMQ基础2

╰半橙微兮° 2022-10-31 04:07 111阅读 0赞

**目录**

1.MQ消息持久化

2.RocketMQ消息存储结构

3.RocketMQ消息数据持久化方式

4.RocketMQ高可用性

5.RocketMQ负载均衡

6.RocketMQ消息重试机制

6.1顺序消息的重试

6.2无序消息的重试

7.死信队列

7.1死信特性

7.2查看死信信息

8.消息的幂等性

8.1产生重复消息的场景

8.2处理策略

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

# 1.MQ消息持久化 #

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

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

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

RocketMQ在持久化时：消息用顺序写,保证了消息存储的速度，采用“零拷贝”技术，提高消息存盘和网络发送的速度（采用java的MappedByteBuffer 实现：省去向用户态的内存复制 ，限制条件时文件大小时1.5G-2G）

![20210221090033941.png][]

# 2.RocketMQ消息存储结构 #

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70][]

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

配置文件设置的路径：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 1][]

持久化文件：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 2][]    ![20210221100144828.png][]

![20210221100227564.png][]        ![20210221100254121.png][]

# 3.RocketMQ消息数据持久化方式 #

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 3][]

*  同步刷盘 ：在返回写成功状态时，消息已经被写入磁盘。具体流程是，消息写入内存的PAGECACHE后，立刻通知刷盘线程刷盘， 然后等待刷盘完成，刷盘线程执行完成后唤醒等待的线程，返回消息写 成功的状态。
 *  异步刷盘 ：在返回写成功状态时，消息可能只是被写入了内存的PAGECACHE，写操作的返回快，吞吐量大；当内存里的消息量积累到一定程度时，统一触发写磁盘动作，快速写入。
 *  主从同步复制：
    
    同步复制方式是等Master和Slave均写 成功后才反馈给客户端写成功状态；
    
    在同步复制方式下，如果Master出故障， Slave上有全部的备份数据，容易恢复，但是同步复制会增大数据写入 延迟，降低系统吞吐量。
 *  主从异步复制：
 *  异步复制方式是只要Master写成功 即可反馈给客户端写成功状态。
    
    在异步复制方式下，系统拥有较低的延迟和较高的吞吐量，但是如果Master出了故障，有些数据因为没有被写 入Slave，有可能会丢失；

刷盘配置：flushDiskType ：SYNC\_FLUSH、ASYNC\_FLUSH

主从配置：brokerRole ：ASYNC\_MASTER（主节点）、 SYNC\_MASTER（主节点）、SLAVE（从节点）

*通常使用异步刷盘+主从同步复制：*即使有一台 机器出故障，仍然能保证数据不丢

# 4.RocketMQ高可用性 #

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 4][]

Nameservice 和 broker均可采用集群部署

*  消息消费高可用 ：在Consumer的配置文件中，并不需要设置是从Master读还是从Slave 读，当Master不可用或者繁忙的时候，Consumer会被自动切换到从Slave 读。有了自动切换Consumer这种机制，当一个Master角色的机器出现故障后，Consumer仍然可以从Slave读取消息，不影响Consumer程序。这就达到了消费端的高可用性。
 *  消息发送高可用 ：在创建Topic的时候，把Topic的多个Message Queue创建在多个Broker组上（相同Broker名称，不同 brokerId的机器组成一个Broker组），这样当一个Broker组的Master不可 用后，其他组的Master仍然可用，Producer仍然可以发送消息。 RocketMQ目前还不支持把Slave自动转成Master，如果机器资源不足， 需要把Slave转成Master，则要手动停止Slave角色的Broker，更改配置文 件，用新的配置文件启动Broker。

# 5.RocketMQ负载均衡 #

Producer端：每个实例在发消息的时候，默认会轮询所有的message queue发送，以达到让消息平均落在不同的queue上。而由于queue可以散落在不同的broker，所以消息就发送到不同的broker下。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 5][]

Consumer端：

默认：AllocateMessageQueueAveragely：将所有的实例队列平均分给所有的消费者，如果消费者发生碧昂更，重新出发算法。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 6][]

AllocateMessageQueueAveragelyByCircle：也是平均分摊每一条queue，只是以环状轮流分queue的形式

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 7][]

*广播模式不涉及负载均衡*

# *6.*RocketMQ消息重试机制 #

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

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

## 6.2**无序消息的重试** ##

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

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

（1）重试次数

消息队列 RocketMQ 默认允许每条消息最多重试 16 次（条消息无论重试多少次，这些重试消息的 Message ID 不会改变 ），如果消息重试 16 次后仍然失败，消息将不再投递。如果严格按照上述重试时间间隔计算，某条消息在一直消费失败的前提下，将会在接下来的 4 小时 46 分钟之内进行 16 次重试，超过这个时间范围消息将不再重试投递。 每次重试的间隔时间如下：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 8][]

（2）消费者触发重试

集群消费方式下，消息消费失败后期望消息重试，需要在消息监听器接口的实现中明确进行配置（三种方式任选一种）：

*  返回 Action.ReconsumeLater （推荐）
 *  返回 Null
 *  抛出异常

public class MessageListenerImpl implements MessageListener {
        @Override
        public Action consume(Message message, ConsumeContext context) {
            //处理消息
            doConsumeMessage(message);
            //方式1：返回 Action.ReconsumeLater，消息将重试
            return Action.ReconsumeLater;
            //方式2：返回 null，消息将重试
            return null;
            //方式3：直接抛出异常， 消息将重试
            throw new RuntimeException("Consumer Message exceotion");
        }
    }

（3）消费失败后，不重试配置方式

集群消费方式下，消息失败后期望消息不重试，需要捕获消费逻辑中可能抛出的异常，最终返回 Action.CommitMessage，此后这条消息将不会再重试。

public class MessageListenerImpl implements MessageListener {
        @Override
        public Action consume(Message message, ConsumeContext context) {
            try {
                doConsumeMessage(message);
            } catch (Throwable e) {
                //捕获消费逻辑中的所有异常，并返回 Action.CommitMessage;
                return Action.CommitMessage;
            }
            //消息处理正常，直接返回 Action.CommitMessage;
            return Action.CommitMessage;
        }
    }

（4）自定义消息最大重试次数

消息队列 RocketMQ 允许 Consumer 启动的时候设置最大重试次数，重试时间间隔将按照如下策略：

*  最大重试次数小于等于 16 次，则重试时间间隔同上表描述。
 *  最大重试次数大于 16 次，超过 16 次的重试时间间隔均为每次 2 小时。

*  注意：*

*  *消息最大重试次数的设置对相同 Group ID 下的所有 Consumer 实例有效。*
 *  *如果只对相同 Group ID 下两个 Consumer 实例中的其中一个设置了 MaxReconsumeTimes，那么该配置对两个 Consumer 实例均生效。*
 *  *配置采用覆盖的方式生效，即最后启动的 Consumer 实例会覆盖之前的启动实例的配置*

Properties properties = new Properties();
    //配置对应 Group ID 的最大消息重试次数为 20 次
    properties.put(PropertyKeyConst.MaxReconsumeTimes,"20");
    Consumer consumer =ONSFactory.createConsumer(properties);

（5）获取重试次数

消费者收到消息后，可按照如下方式获取消息的重试次数：

public class MessageListenerImpl implements MessageListener {
        @Override
        public Action consume(Message message, ConsumeContext context) {
            //获取消息的重试次数
            System.out.println(message.getReconsumeTimes());
            return Action.CommitMessage;
        }
    }

# 7.死信队列 #

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

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

## 7.1死信特性 ##

死信消息具有以下特性

*  不会再被消费者正常消费。
 *  有效期与正常消息相同，均为 3 天，3 天后会被自动删除。因此，请在死信消息产生后的 3 天内及时处理。

死信队列具有以下特性：

*  一个死信队列对应一个 Group ID， 而不是对应单个消费者实例。
 *  如果一个 Group ID 未产生死信消息，消息队列 RocketMQ 不会为其创建相应的死信队列。
 *  一个死信队列包含了对应 Group ID 产生的所有死信消息，不论该消息属于哪个 Topic。

## 7.2查看死信信息 ##

（1）在控制台查询出现死信队列的主题信息

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 9][]

（2）在消息界面根据主题查询死信消息

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 10][]

# 8.消息的幂等性 #

消息在消费方，对于同一条消息不管消费了多少次，结果都是一样的，因此有必要根据业务上的唯一 Key 对消息做幂等处理的必要性 ）。

## 8.1产生重复消息的场景 ##

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

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

## 8.2处理策略 ##

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

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

consumer.subscribe("ons_test", "*", new MessageListener() {
        public Action consume(Message message, ConsumeContext context) {
            String key = message.getKey()
            // 根据业务唯一标识的 key 做幂等处理
        }
    });

[20210221090033941.png]: /images/20221024/5922e10db831419b8b5da03c5838ff9b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70]: /images/20221024/16ecdb0452034102bbbefff815d39256.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 1]: /images/20221024/789d4a690dd94edead2a2e9204845eec.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20210221100104302.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3,size_16,color_FFFFFF,t_70
[20210221100144828.png]: /images/20221024/7c30e6f858ea4a7d89294686488ef897.png
[20210221100227564.png]: https://img-blog.csdnimg.cn/20210221100227564.png
[20210221100254121.png]: /images/20221024/76b5c5607de54b7d822a3a03429b06e7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 3]: /images/20221024/c4dc4e55c2a1411ca008aa91324d5b49.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 4]: /images/20221024/a0364e939b714c649c6acf3138853d83.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 5]: /images/20221024/bd4320840e414c4cbc924585dae0830f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 6]: /images/20221024/d7c54c76b27c40c7b1640296c14a12b2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 7]: /images/20221024/af371c1d5df14d69bbc12223a1042045.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 8]: /images/20221024/fc5228a130cb4b36845bcade3cb8d7cd.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 9]: /images/20221024/dcd08f81af2d4815837039fafb1e8480.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2MTU0MDc3_size_16_color_FFFFFF_t_70 10]: /images/20221024/3e63f4d9c4244e4e964966f180f05fd3.png