【RocketMQ】原理分析：Consumer消费模型、消费规则

爱被打了一巴掌 2022-12-22 08:38 274阅读 0赞

## 1.RocketMQ消费模型 ##

RocketMQ提供了两种消息消费模型，一种是pull主动拉取，另一种是push被动接收。

*  实际上 RocketMQ都是pull模式，而push只是在pull模式上做了一层封装
 *  RocketMQ是基于长轮训来实现消息的pull
 *  pull到消息以后触发业务消费者注册的call back

## 2.消费规则 ##

我们在[【RocketMQ】原理分析：Broker集群模式、队列分区][RocketMQ_Broker] 介绍了RocketMQ的多住多从集群模式和通过多个消息队列实现分区。所以如果有一个consumer来了，他到底该在哪个小集群去消费topic下的哪个消息队列？

### 2.1 master角度 ###

首先，站在master角度看看Consumer是如何分配拉取消息的。假设有两个master节点的集群，分别是A，B一个消费者组，组中有2个consumer。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center]

因为master集群数：group中消费者数 = 1:1，所以两个master节点恰好可以平均分发到两个消费者上（比如consumer1消费brokerA的消息，consumer2消费brokerB的消息）

另外，如果此时只有一个Consumer，那么它会消费两个master节点的数据；如果此时有3个Consumer，那么会有一个Consumer无法消费到消息

### 2.2 topic角度 ###

接下来，站在topic角度看看Consumer是如何分配拉取消息的。假设有两个master节点的集群，创建了一个TestTopic，并且对这个topic创建了两个队列，也就是分区。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center 1]  
一个消费者可以同时消费多个master节点上的消息，而Topic中的队列数 ：group中的消费者数 = 1:1，所以无论是在哪个group都是一个consumer消费Broker集群中一个队列（比如consumer1消费MQ0，consumer2消费MQ1）

> 注：消费者定义了两个分组，分组的概念也是和kafka一样，通过分组可以实现消息的广播。每个group中有2个Consumer

另外，如果此时只有一个Consumer，那么它会消费所有分区；如果此时有3个分区，那么就有一个Consumer消费2个分区。

### 2.3 动态平衡 ###

数量关系确定了，那么具体的对应关系是什么呢？比如consumer1到底是消费MQ0还是MQ2。

和kafka一样，消费端也会针对Message Queue做**负载均衡**，使得每个消费者能够合理的消费多个分区的消息。消费端会通过Rebalance Service线程，10秒钟做一次基于topic下的所有队列负载消费端遍历自己的所有topic，

1.  依次调rebalanceByTopic 根据topic获取此topic下的所有queue
2.  选择一台broker获取基于group的所有消费端（有心跳向所有broker注册客户端信息）
3.  选择队列分配策略实例AllocateMessageQueueStrategy执行分配算法（RocketMQ提供了6中分区的分配算法）
    
     *  AllocateMessageQueueAveragely：平均分配算法（默认）
     *  AllocateMessageQueueAveragelyByCircle：环状分配消息队列
     *  AllocateMessageQueueByConﬁg：按照配置来分配队列： 根据用户指定的配置来进行负载
     *  AllocateMessageQueueByMachineRoom：按照指定机房来配置队列
     *  AllocateMachineRoomNearby：按照就近机房来配置队列：
     *  AllocateMessageQueueConsistentHash：一致性hash，根据消费者的cid进行

什么时候触发负载均衡？消费者启动之后消费者数量发生变更 每10秒会触发检查一次rebalance

## 3.消息的顺序消费 ##

消费消息的顺序要同发送消息的顺序一致。举个例子，一个订单产生了3条消息，分别是订单创建，订单付款，订单完成。消费时，要按照这个顺序消费才能有意义。

保证顺序消费的核心就在于：producer - queue - consumer 一对一关系

1）Producer 发送消息的消息经过分发后还是到同一个 queue => producer 自定义分发规则

for (int i = 0; i < 100; i++) {
        
    
        // 订单ID相同的消息要有序
        int orderId = i % 10;
        Message msg = new Message("TopicTest", tags[i % tags.length], "KEY" + i, ("Hello " + i).getBytes());
    
        SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
        
            @Override
          	// 重写 select 方法，自定义分发规则
            public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
        
                Integer id = (Integer) arg;
              	// queueIdx = 订单 ID % queue 的数量
                int index = id % mqs.size();
                return mqs.get(index);
            }
        }, orderId);
    
        System.out.println(sendResult);
    }

2）一个 queue 只能由一个 consumer 去消费 => MessageListenerOrderly（特性：一个队列只会被一个线程取到）

这里再说一下，MetaQ 对每个Consumer提供了两种消费方式：

*  MessageListenerConcurrently：并发监听，也就是基于多个线程并行来消费消息。这个无法保证消息消费的顺序
 *  MessageListenerOrderly：顺序消费，单线程去消费消息

consumer.registerMessageListener(new MessageListenerOrderly() {
        
        @Override
        public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
        
          	msgs.stream().forEach(messageExt -> System.out.println(new String(messageExt.getBody())));
            return ConsumeOrderlyStatus.SUCCESS;
        }
    });

3）最后，一个消费者内部对一个mq的消费要保证是有序的。

顺序消费会带来一些问题，

1.  遇到消息失败的消息，无法跳过，当前队列消费暂停
2.  降低了消息处理的性能

[RocketMQ_Broker]: https://blog.csdn.net/weixin_43935927/article/details/109775265
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/b8f8fb5f92114d58affd6c3945b6f846.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221120/946d052cc9c749bfa5b0935adaa6c3a1.png