透彻理解 RocketMQ
首页 死磕 Java 优质系列 Java 技术总览 Java 面试宝典 关于我

分布式消息中间件（八）——RocketMQ消费者原理

转
 2023-08-01
©版权
原文作者：Ressmix 原文地址：https://www.tpvlog.com/article/126

在RocketMQ持久化原理中，我们已经讲解了生产者（Producer）发送消息的基本原理。本章，我们来看看消费者（Consumer）消费消息的基本原理。

一、消费者组

在RabbitMQ中，每一个消费者实例，都必须属于某个Group ID——群组ID，相当于用一个Group ID把一群Consumer实例归为了一类。这里需要特别注意的是： 同一个消费者 Group ID 下所有 Consumer 实例，订阅关系必须完全一致。如果订阅关系不一致，消息消费的逻辑很容易出现混乱，可能导致消息丢失，甚至出现其它各种莫名其妙的问题！

1.1 订阅关系

什么是订阅关系？我们知道，消息肯定要有其所属的Topic分类，所以订阅关系一致就是指：

  1. 订阅的 Topic 必须一致；
  2. 订阅的 Topic 中的 Tag 必须一致。

我们先来看个正确的订阅关系示例：

202308012101037911.png

从上图可以看到，同一个消费者群组中的所有消费者实例，Topic+Tag是完全一致的。

我们再来看个错误的订阅关系示例：

202308012101046742.png

我们可以在代码中通过下面的方式设置消费者的群组、Topic、Tag：

    // 设置消费者群组：wms_group
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("wms_group");
    
    // 设置Topic：order_topic，Tag：TagA
    consumer.subscribe("order_topic", "TagA");

二、消费模式

RabbitMQ支持两种消费模式： 集群消费模式 、 广播消费模式 。

2.1 集群消费模式

所谓集群，就是指使用了相同 Group ID 的消费者，它们属于同一个集群。 当使用集群消费模式时，RocketMQ认为一条消息只需要被集群内的任意一个消费者处理即可。

集群消费模式，一般适用于每条消息只需要被处理一次的场景，消费进度在服务端维护，可靠性更高，也是默认的模式：

202308012101051403.png

上图中，其实只有一个Group ID 1群组，这个群组中的Consumer实例分布在三台不同的机器上，同时这个群组订阅了Topic+Tag（比如："order_topic", "TagA"），所以对于Broker中的每一条相关消息，只能被这个群组中的某个Consumer实例消费到。

注意：集群消费模式下，不保证每一次失败重投的消息路由到同一台机器上。

2.2 广播消费模式

当使用广播消费模式时，RocketMQ会将一条消息推送给集群内的所有消费者，保证消息至少被每个消费者消费一次。

广播消费模式，一般适用于每条消息需要被集群下的每个消费者处理的场景，消费进度在客户端维护，出现重复消费的概率稍大于集群模式：

202308012101057314.png

上图中，只有一个Group ID 1群组，这个群组中的Consumer实例分布在三台不同的机器上，同时这个群组订阅了Topic+Tag（比如："order_topic", "TagA"），对于Broker中的每一条相关消息，Group ID 1群组中的每一个消费者实例都能消费到该消息。

可以通过以下方式开启消费者实例的广播消费模式：

    consumer.setMessageModel(MessageModel.BROADCASTING); //设置广播消费模式

2.3 使用集群模式模拟广播

上面的两个示例都是只有一个群组，这个群组内的Consumer实例都订阅了相同的Topic+Tag。如果我们有多个不同的群组呢？比如下面这种情况：

202308012101063295.png

上图中，一共有Group ID 1、Group ID 2、Group ID 3三个群组，每个群组都订阅了相同的Topic+Tag（比如："order_topic", "TagA"），这种情况下，对于Broker中的每一条相关消息，三个群组都能消费到该消息，与此同时，这条消息又只能被某个群组中的一个Consumer实例消费到。

比如对于消息MessageN，Group ID 1、Group ID 2、Group ID 3都能消费到这条消息，相当于MessageN拷贝了三份，但是从组内来看，比如Group ID 1，组内只有一个Consumer实例能消费到MessageN。

三、消费方式

3.1 Pull/Push

消息者有两种方式从Broker中的MessageQueue获取消息： Pull方式 和 Push方式 。这两种方式的本质是一样的：都是消费者实例主动发送请求到Broker中拉取消息。Push方式的消息时效性更好一些。

Push方式也只是名字叫Push而已，并不是Broker真的会主动把消息推给Consumer，而是当Consumer主动去获取消息时，如果有新的消息可以消费，那么Broker会立马返回一批消息给Consumer，Consumer处理完后会接着发送请求到Broker拉取下一批消息，这样看起来就好像是Broker在不断推消息给Consumer。

此外，在Push方式下，具有请求挂起/长轮询机制。

请求挂起/长轮询：
当Consumer的请求发送到Broker，如果Broker发现没有新的消息可供消费时，就会让请求线程挂起，默认15秒，在此期间，Broker有后台线程每隔一会儿去检查一下是否有新的消息给Consumer，如果有新的消息到达就会主动唤醒挂起的线程，然后把消息返回给它。

四、底层原理

我们已经从逻辑上了解了消费者的消费模式及消费方式，那么其底层到底是什么样的呢？根据之前学习到的知识，我们知道：

  1. 一个Topic中的多个MessageQueue会分散在多个Master-Broker上（Slave会去同步数据）；
  2. 每个Broker机器上的一个MessageQueue，对应一个ConsumeQueue（在物理磁盘上其实是对应了多个ConsumeQueue文件）。
  3. Consumer会从NameServer拉取路由信息，所以它知道自己订阅的Topic中MessageQueue分布在哪些Broker上。

所以对于一个Topic，比如我们的order_topic，分了4个MessageQueue，均匀分布在两台机器上，假设我们现在有一个库存消费者群组——wms_group：

  • 如果组内有2个Consumer实例，那么默认情况下可能说就是每个Consumer实例负责2个MessageQueue的读取；
  • 如果组内有5个Consumer实例，那么每个Consumer实例负责1个MessageQueue的读取，还剩一个空闲的。

也就是说， 集群模式下，一个Topic中的多个MessageQueue会均匀分摊给同一消费组内的多个Consumer实例去消费，这里的一个原则就是：集群模式下，同一消费者组内，一个MessageQueue只能被一个Consumer实例处理，但是一个Consumer实例可以负责多个MessageQueue的消息处理。

是否均匀分摊需要视Consumer的消费策略来定，默认情况下就是均匀分摊，后面我们讲Consumer源码时再具体讲其它情况。另外，如果多个消费者群组订阅同一个Topic+Tag，从外部看，其实属于广播模式，上面的原则是不适用的。

4.1 消费进度

当消费者实例指定对某个MessageQueue进行消费时，请求到达Broker后，如果是首次消费，Broker就从这个MessageQueue对应的ConsumeQueue文件中，找到第一条消息的地址，然后去CommitLog中根据这个offset地址读取出消息数据，最后将这条消息数据返回给消费者实例。

所以，消费消息的本质是：根据要消费的MessageQueue以及消费位置，去找到对应的ConsumeQueue，读取里面的消息物理offset偏移量，然后到CommitLog中根据offset读取消息数据，返回给消费者。 当消费者处理完一批消息后，会提交一个消费进度到Broker上去，然后Broker就会存储我们的消费进度，以便下次消费使用 。

4.2 消费负载

之前我们提到过，消息者既可以从Master-Broker拉取消息，也可以从Slave-Broker拉取， 那到底什么时候从Master-Broker拉取，什么时候从Slave-Broker拉取呢？

首先，我们要知道，消费者拉取消息时，Broker会去频繁的去读取ConsumeQueue文件，获取offset。之前RocketMQ持久化原理一章我们讲过，RocketMQ会利用PageCache对磁盘文件的读写进行优化。也就是说，ConsumeQueue文件的内容会被预热到PageCache中，因为ConsumeQueue文件的大小只有几MB，所以Broker读取ConsumeQueue文件的内容时，效率是非常高的，几乎都是直接从内存读取。

但是这里注意：Broker读取完ConsumeQueue中的offset后，还要去CommitLog中读取消息内容，CommitLog文件是比较大的，所以是无法把全部数据都放到PageCache中的，只有那些最新写入的消息，一般才会停留在PageCache中，那些比较老的数据，会被OS异步刷到磁盘上。所以，当Broker从CommitLog中读取消息内容时，就有两种可能：

  1. 读取的是那种刚刚写入CommitLog的消息，此时它们大概率还停留在PageCache中，那么就直接是内存读取，性能是很高的；
  2. 读取的是比较早之前写入CommitLog的数据，此时它们早就被刷入磁盘了，已经不在PageCache中，那么就只能从磁盘上的文件里读取了，这个性能是比较差的；

所以，结论就很明显了：

  • 如果消费者实例一直快速的在拉取并处理消息，速率跟生产者写入的速率差不多，那么每次拉取几乎都是从PageCache从读取数据，此时消费者大概率都是直接从Master-Broker读取消息；
  • 如果Master-Broker的负载很高，导致拉取消息的速度很慢，或者消费者实例自身的处理消息速度很慢，导致跟不上生产者写入的速率，此时Master-Broker就会通知消费者下一次开始从它的Slave节点去拉消息。

我们举个例子来理解下，假设PacheCache最多缓存5w条消息，现在生产者已经写入到第10w万条消息，而消费者只拉取到第2万条，那此时Master-Broker就会认为：你消费者还有8w条消息没消费掉，而我的PageCache最多只能容纳5w条，按照你目前的消费速率，大概率需要我从磁盘上加载数据，所以你下次直接从Slave去拉消息吧，不要对我的性能造成影响。

五、总结

本章，我们对消费者Consumer的基本原理进行了讲解，主要包含三种消费模式、两种消费方式，以及消费者消费消息时底层原理是怎么样的。关于消费者（Consumer）的更多细节点和使用最佳实践，读者可以参考官方的文档。

阅读全文
文章标签： 消息队列RocketMQ