21 消费者获取消息处理以及进行ACK的原理

绝地灬酷狼 2022-10-10 06:03 13阅读 0赞

### 1.消费组概念 ###

消费者组的意思，就是让你给一组消费者起一个名字。

**场景解释消费者组**

当前有个Topic为“TopicOrderPaySuccess”，此时有库存系统、营销系统等要去消费这个Topic中的数据。这里将库存系统命名为“stock\_consumer\_group”。

设置消费者组的方式在代码里是这样的：

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("stock_consumer_group");

假设该库存系统部署了4台机器，每台机器上的消费者组的名字都是“stock\_consumer\_group”，那么这4台机器就同属于一个消费者组，以此类推，每个系统的机台机器都是属于各自的消费者组的。

**消费者组如何消费消息**

当前场景中，库存系统、营销系统各有两台机器，各自在两台机器上都部署了消费者组，都订阅了“TopicOrderPaySuccess”这个Topic的消息。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQwNTY3MA_size_16_color_FFFFFF_t_70][]

此时，当订单系统为生产者发送了一条消息到这个Topic上，此时库存系统、营销系统都回去拉取这条消息。这里要说明的是，库存系统、营销系统的各自的两台机器**只有一台机器会获取到这个消息**。

小结：不同的系统应该设置不同的消费者组，如果不同的消费组都订阅了同一个Topic，对Topic里的一条消息，每个消费者都会获取到这条消息。在正常情况下，每个消费组只有一台机器获取到一次消息。

### 2.集群模式消费 vs 广播模式消费 ###

**集群模式**

前面的案例中，使用的是集群模式，也就是说，一个消费组获取到一条消息，只会交给组内的一台机器去处理，不是每台机器都可以获取到这条消息的。

**广播模式**

代码如下：

consumer.setMessageMode(MessageModel.BROADCASTING);

如果修改为广播模式，那么对于消费组来说，组内每台机器都可以获取到这条消息。

一般情况下，广播模式很少使用，基本上都是使用集群模式来进行消费的。

### 3.MessageQueue与消费者的关系 ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQwNTY3MA_size_16_color_FFFFFF_t_70 1][]

对于Topic的各个MessageQueue而言，就是通过各个ConsumeQueue文件来存储属于MessageQueue的消息在CommitLog文件中的物理地址，也就是一个offset偏移量。

而在**消费时，**一个Topic的多个MessageQueue会均匀分摊给消费组内的多个机器去消费，这里的一个原则就是，一个MessageQueue只能被一个消费机器去处理，但是一台消费者机器可以负责多个MessageQueue的消息处理。

**场景说明：**

在以下场景中，Topic下有4个MessageQueue，分布于两个Master Broker上，每个Master Broker上有2个MessageQueue。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQwNTY3MA_size_16_color_FFFFFF_t_70 2][]

此时，库存系统作为一个消费组里有两台机器，那么正常情况下，就是让这两台机器每个都负责2个MessageQueue的消费了。

**负载重平衡**

前面的案例中，如果消费者中出现机器宕机或者扩容加机器的情况，一般是如下的处理方案：

这个时候会进入一个rebalance的环节，也就是说重新给各个消费机器分配他们要处理的MessageQueue。

比如：机器01负责MessageQueue01、02，机器02负责MessageQueue03、04；此时如果机器02宕机，那么机器01就会接管机器02之前负责的MessageQueue03、04。

如果此时又新增了一台机器03，那就可以把机器02之前负责的MessageQueue03、04转移给机器03；机器01仅负责原来的01、02。

### 4.Push模式 vs Pull模式 ###

Push和Pull两种模式的本质，都是消费者机器主动发送请求到Broker机器去拉取一批消息下来。

**Push模式**

Push模式下，Broker会尽可能实时的把新消息交给消费者机器来进行处理，它的消息时效性会更好。一般我们使用RocketMQ的时候，消费者模式通常都是基于Push模式来做的。

**Push模式的实现思路**

当消费者发送请求到Broker去拉取消息的时候，如果有新的消息可以消费，那么就会立马返回一批消息到消费机器去处理，处理完之后会接着立刻发送请求到Broker机器去拉取下一批消息。

**Push模式下的请求挂起和长轮询机制**

当你的请求发送到Broker，结果它发现没有新的消息给你处理的时候，就会让请求线程挂起，默认是挂起15秒，然后这个期间它会有后台线程每隔一会儿就去检查一下是否有新的消息给你，另外如果在这个挂起过程中，如果有新的消息到达了会主动唤醒挂起的线程，然后把消息返回给你。

### 5.Broker将消息读取出来返回给消费机器的原理 ###

其原理是，根据你要消费的MessageQueue以及开始消费的位置，去找到对应的ConsumeQueue读取里面对应位置的消息在CommitLog中的物理offset偏移量，然后到CommitLog中根据offset读取消息数据，返回给消费者机器。

### 6.消费者机器如何处理消息、进行ACK以及提交消费进度？ ###

消费者机器在拉取到一批消息之后，就会将这批消息回调我们注册的一个函数，如下：

consumer.registerMessageListener(new MessageListenerConcurrently() {
            
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(
                      List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
            
                // 处理消息
                //标记该消息已经被成功消费
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

当我们处理完这批消息之后，消费者机器就会提交我们目前的一个消费进度到Broker上去，然后Broker就会存储我们的消费进度。

### 7.问题思考 ###

（1）一般我们获取到一批消息之后，什么时候才可以认为是处理完这批消息了？是刚拿到这批消息就算处理完吗？还是说要对这批消息执行完一大堆的数据库之类的操作，才算是处理完了？

答：只有提交了消费进度，才会被认为消费完成。

（2）如果获取到了一批消息，还没处理完呢，结果机器就宕机了，此时会怎么样？这些消息会丢失，再也无法处理了吗？

答：消息不会丢失，消息会重试消费，知道消费成功。

（3）如果获取到了一批消息，已经处理完了，还没来得及提交消费进度，此时机器宕机了，会怎么样呢？

答：消息会重新消费，所以需要自己在业务上做幂等，防止重复消费。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQwNTY3MA_size_16_color_FFFFFF_t_70]: /images/20221005/606b0ec38b9d448a90b864bc575dd374.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQwNTY3MA_size_16_color_FFFFFF_t_70 1]: /images/20221005/affa1f97e1e04679bfb775f112903d93.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQwNTY3MA_size_16_color_FFFFFF_t_70 2]: /images/20221005/72f58d2f32cd451db8d2e4376e0b5d79.png