【RocketMQ】Consumer消息拉取和消费流程分析

MQConsumer是RocketMQ提供的消费者接口，从接口定义上可以看到，它主要的功能是订阅感兴趣的Topic、注册消息监听器、启动生产者开始消费消息。 ​

消费者获取消息的模式有两种：推模式和拉模式，对应的类分别是DefaultMQPushConsumer和DefaultMQPullConsumer，需要注意的是，在4.9.0版本，DefaultMQPullConsumer已经被废弃了。

Push模式下，由Broker接收到消息后主动推送给消费者，实时性较高，但是会增加Broker的压力。Pull模式下，由消费者主动从Broker拉取消息，主动权在消费者，这种方式更灵活，消费者可以根据自己的消费能力拉取适量的消息。 ​

实际上，Push模式也是通过Pull的方式实现的，消息统一由消费者主动拉取，那如何保证消息的实时性呢？ ​

Consumer和Broker会建立长连接，一旦分配到MessageQueue，就会立马构建PullRequest去拉取消息，在不触发流控的情况下，不管有没有拉取到新的消息，Consumer都会立即再次拉取，这样就保证了消息消费的实时性。 ​

如果Broker长时间没有新的消息，Consumer一直拉取，岂不是空转CPU浪费资源？ ​

Consumer在拉取消息时，会携带参数suspendTimeoutMillis，它表示Broker在没有新的消息时，阻塞等待的时间，默认是15秒。如果没有消息，Broker等待15秒再返回结果，避免客户端频繁拉取。如果15秒内有新的消息了，立马返回，保证消息消费的时效性。 ​

本文将重点分析DefaultMQPushConsumer，看看Push模式下，消费者是如何启动的、消息是如何拉取的、又是如何消费的。 ​

在直接看代码之前，先来大概了解一下Consumer涉及的相关组件，是很有必要的。

RocketMQ暴露给开发者使用的基于Push模式的默认生产者类，和DefaultMQProducer一样，它也仅仅是一个外观类，基本没有业务逻辑，几乎所有操作都转交给生产者实现类DefaultMQPushConsumerImpl完成。这么做的好处是RocketMQ屏蔽了内部实现，方便在后续的版本中随时更换实现类，而用户无感知。 ​

默认的基于Push模式的消费者实现类，拥有消费者的所有功能，例如：拉取消息、执行钩子函数、消费者重平衡等等。 ​

调用拉取消息API的包装类，它是Consumer拉取消息的核心类，它有一个方法特别重要pullKernelImpl，是拉取消息的核心方法。它会根据拉取的MessageQueue去查找对应的Broker，然后构建拉取消息请求头PullMessageRequestHeader发送到Broker，然后执行拉取回调，在回调里会通知消费者消费拉取到的消息。 ​

OffsetStore是RocketMQ提供的，用来帮助Consumer管理消费位点(消费进度)的接口，它有两个实现类：LocalFileOffsetStore和RemoteBrokerOffsetStore，从名字就可以看出来，一个是将消费进度存储在本地，一个是将消费进度存储在Broker上。 LocalFileOffsetStore会将消费进度持久化到本地磁盘，Consumer启动后会从指定目录读取文件，恢复消费进度。 RemoteBrokerOffsetStore将消费进度交给Broker管理，Consumer不会存储到文件，没有意义，但是消费消息时会暂存消费进度在内存，然后在拉取消息时上报消费进度，由Broker负责存储。

什么场景下需要将消费进度存储在本地呢？这和RocketMQ消息消费模式有关，RocketMQ支持两种消息消费模式：集群消费和广播消费。一个ConsumerGroup下可以有多个消费者实例，集群模式下，消息只会投递给其中一个Consumer实例消费，而广播模式下，消息会投递给每个Consumer实例。 ​

集群模式消费 广播模式消费 综上所述，集群模式下，消费进度由Broker管理，使用RemoteBrokerOffsetStore。广播模式下，因为消息需要被每个Consumer实例消费，每个实例消费的进度是不一样的，因此由实例自己存储消费进度，使用LocalFileOffsetStore。 ​

消费消息的服务，客户端拉取到消息后，是需要有线程去消费的，因此它是一个线程池，线程数由consumeThreadMin和consumeThreadMax设置，默认线程数为20。 ​

它是一个接口，比较重要的两个方法如下：

一个是由当前线程直接消费消息，另一个是提交消费请求ConsumeRequest由线程池去负责调度，一般情况下使用的还是后者。 ​

RocketMQ提供了两个实现类，分别是ConsumeMessageConcurrentlyService和ConsumeMessageOrderlyService，前者用来并发消费消息，后者用来消费有序消息，本文只分析前者。 ​

消息拉取服务，负责从Broker拉取消息，然后提交给ConsumeMessageService消费。它也是一个线程，它的run方法是一个死循环，通过监听阻塞队列来判断是否需要拉取消息。阻塞队列里存放的就是PullRequest对象，当Consumer实例上线后，会做一次负载均衡，从众多MessageQueue中给自己完成分配，当有新的MessageQueue被分配给自己，就会创建PullRequest对象提交到阻塞队列，然后PullMessageService就会开始拉取消息，在拉取完成的回调函数中，不管有没有拉取到新的消息，在不触发流控的情况下，都会一直拉取。 ​

消费者实现比生产者实现要复杂的多，笔者也是研究了很久，才清楚了大体流程。整个Consumer的运行可以大致分为四个过程：

下面我们一步步来，各位看官，请细听分说。

Consumer的启动流程和Producer有部分重合之处。 ​

首先自然是创建DefaultMQPushConsumer，在它的构造函数中，会创建消费者实现类DefaultMQPushConsumerImpl。

前面说过，DefaultMQPushConsumer只是一个外观类，它更多的职责只是保存Consumer的配置，它的属性可以重点关注一下：

Consumer要订阅自己感兴趣的Topic，支持通过子表达式过滤消息，子表达式的类型有两种：TAG和SQL92。

不管是哪种表达式，最终都会被解析成订阅关系SubscriptionData对象存储到Map中。

SubscriptionData代表了消费者的订阅关系，属性如下：

如果使用TAG的方式，会计算出Tag哈希值，Broker在ConsumeQueue索引中记录了Tag哈希，这样就可以根据Tag哈希快速过滤消息了。 ​

订阅完Topic，就是注册消息监听MessageListener，就是一个赋值操作，跳过。 ​

以上操作执行完，Consumer就可以启动了，接下来才是重头戏。 ​

1.外观类启动的时候，会启动消费者实现类DefaultMQPushConsumerImpl，我们直接看它就好。启动主要做了以下事情：

​

2.checkConfig方法，会在Consumer启动前做一系列的校验，确保服务满足启动条件，校验的事项有：

启动前校验通过，说明配置没有问题，具备启动的基本条件。 ​

3.copySubscription方法会拷贝订阅关系到RebalanceImpl，Consumer在重平衡时需要用到，除了拷贝给定的Topic订阅关系，Consumer还会自动订阅ConsumerGroup的重试队列。

​

4.创建客户端实例MQClientInstance，消息拉取核心对象PullAPIWrapper。

根据消息消费模式，创建对应的OffsetStore。

创建ConsumeMessageService并启动，如果是有序消息，创建ConsumeMessageOrderlyService，并发消费创建ConsumeMessageConcurrentlyService。

ConsumeMessageService是一个线程池，消息拉取服务拉取到消息后，会构建ConsumeRequest对象交给线程池调度执行。

5.前置操作完成后，就可以启动客户端实例了。MQClientInstance启动主要做了以下事情：

客户端如果没有指定NameServer地址，RocketMQ会读取环境变量rocketmq.namesrv.domain，它的期望值是一个URL链接，每隔2分钟发一个请求更新NameServer地址。在集群环境下，NameServer机器数和IP都是不固定的，通过配置中心下发比硬编码更灵活。

RocketMQ基于Netty来完成网络通信，Consumer作为客户端是要和Broker通信的，因此还需要启动Netty客户端。

启动各种定时任务，这些任务包括：获取NameServer地址，从NameServer拉取Topic路由信息、清理下线的Broker、给Broker发心跳、持久化消费进度。 ​

启动消息拉取服务PullMessageService，它是一个单独的线程，run方法会监听阻塞队列pullRequestQueue，只要队列中有拉取请求，它就会去Broker拉取消息。

​

启动重平衡服务RebalanceService，也是一个单独的线程，默认会每隔20秒重新做一次负载均衡，给Consumer重新分配MessageQueue。例如，TopicA下有4个MessageQueue，此时只有一个消费者实例订阅了，那么这4个MessageQueue都会分配给它消费。过了一会儿，新的消费者实例上线，此时会做一次重平衡，重新分配，因为有两个消费者实例了，因此每个实例会分配2个MessageQueue。

6.相关服务启动完成后，Consumer会自动向NameServer拉取订阅的Topic路由信息。

如果订阅子表达式用的是SQL92的语法，还需要将表达式上传到Broker进行编译，对应的方法是：

新的Consumer实例启动上线了，会向Broker发送心跳，Broker接收到心跳后，会发命令NOTIFY_CONSUMER_IDS_CHANGED给Group下其它消费者，要求它们重新做一次负载均衡，重新分配MessageQueue。

Consumer的心跳除了告诉Broker自己还活着，还做了一些其他事情：

Broker通知其它Consumer重平衡，当前实例自身也要重平衡呀，执行方法：

重平衡里做了非常多的事情，其中就包括主动拉取新分配的MessageQueue，这个后面再说。 ​

至此，Consumer就算启动完成了。

在RocketMQ的设计理念中，不管是Producer往队列里发消息，还是Consumer从队列里消费消息，负载均衡都是在客户端完成的。 ​

有三种情况会触发Consumer进行重平衡：

重平衡的入口方法为MQClientInstance.doRebalance()，拿第三种情况为例，它的run方法如下：

Consumer实例启动后，自身会主动进行一次重平衡，时序图如下：

客户端重平衡时，会遍历所有Consumer，根据订阅的Topic进行重平衡，所谓的「重平衡」就是将Topic下的MessageQueue重新分配给Consumer消费，分配的算法策略对应接口AllocateMessageQueueStrategy。 ​

根据Topic重平衡的流程是这样的，先获取Topic下所有的MessageQueue，然后获取订阅Topic的所有客户端ID集合，对队列集合和客户端集合排序，根据分配策略给当前Consumer分配MessageQueue。

allocateResult就是新的MessageQueue分配结果，分配的MessageQueue有可能变多、变少或者不变。变多的话，会创建PullRequest去拉取MessageQueue里的消息。变少就移除MessageQueue，然后将对应的ProcessQueue挂起，消息停止消费，上报消费位点。这是Consumer启动时主动触发的重平衡操作。 ​

新的Consumer上线，其它Consumer需要将自己负责消费的MessageQueue分配一点给它。旧的Consumer下线，分配给它的MessageQueue无法被消费了，需要重新分配给其它Consumer。这两种情况，Broker都要及时通知Group下其它所有Consumer执行重平衡服务。 ​

新的Consumer上线是通过心跳的方式通知Broker的，旧的Consumer下线是Broker通过监听Channel关闭事件来感知的，无论何种方式，Broker的处理方式都是一样的，给Group下所有Consumer发送命令NOTIFY_CONSUMER_IDS_CHANGED，Consumer接受到Broker的请求后，会执行重平衡服务。

这里以心跳为例，Broker在收到Consumer的心跳包后，反序列化为HeartbeatData对象，然后将客户端相关信息封装为ClientChannelInfo对象。

然后向ConsumerManager注册消费者实例，注册的时候会进行判断，是否是新的消费者实例，消费者的订阅关系是否发生变化，两个条件满足其一都要通知Consumer重平衡。

遍历所有的Consumer客户端Channel，通过Broker2Client给客户端发送请求。

Broker通知Consumer重平衡的方法是notifyConsumerIdsChanged，逻辑很简单，构建对应的请求头，通过Netty发送数据包。

Consumer在接收到Broker的请求后，会通过ClientRemotingProcessor类处理请求，其实就是调用了rebalanceImmediately方法，代码如下：

这是Broker主动通知Consumer重平衡的全流程。

前面已经说过，新的Consumer实例启动后，它会主动执行一次重平衡操作，给自己分配MessageQueue。此时，它的队列集合是空的，因此被分配到的MessageQueue一定是新的，那么它就会构建PullRequest去拉取这些新分配的MessageQueue里的消息。

Consumer消息拉取的时序图：

构建的PullRequest对象会被分发到PullMessageService的阻塞队列里，此时消息拉取线程会被唤醒，然后执行pullMessage方法。

PullRequest属性如下：

PullRequest会转交给消费者实现类处理，在拉取消息之前，进行必要的流控处理，流控是用来保护消费者的，当消费者消费能力不够时，拉取速度太快会导致大量消息积压，很可能内存溢出。以下情况会被流控：

一旦触发流控，默认会过50ms再尝试拉取，这个时间是可配置的。 ​

没有触发流控，则开始消息拉取，Consumer在拉取消息时，会顺便上报消费位点CommitOffset。

拉取消息的核心方法是PullAPIWrapper的pullKernelImpl方法。它首先会查找拉取的MessageQueue所在Broker的Master机器地址。

然后构建消息拉取请求头PullMessageRequestHeader，下面是请求头的属性：

queueId告诉Broker自己要从哪个MessageQueue拉取消息，queueOffset是消息拉取的偏移量，commitOffset是上报的消费位点，subscription是消息过滤的子表达式，可以是TAG，也可以是SQL92语法。

请求头构建完毕，接下来就是通过Netty客户端将请求发送到Broker了，Broker会检索消息然后返回给Consumer，消息拉取流程到此结束。

Consumer拉取到消息后，自然就是消费了，何时会触发消息的消费动作呢？ ​

在消息拉取时，会创建PullCallback对象，消息拉取完成后会触发对应的回调，在回调方法里如果拉取到了消息会触发消息消费流程。 ​

Consumer消息消费的时序图：

在PullCallback的onSuccess方法中，会对消息拉取结果进行判断，PullStatus.FOUND代表拉取到了新的消息，此时会提交消费请求到ConsumeMessageService，消费者线程开始消费消息。

在submitConsumeRequest方法中，如果拉取到的消息数超过了Consumer单次消费的最大消息数，就会对消息进行切分，封装成多个ConsumeRequest处理，否则就直接全部处理。