深入研究RocketMQ消费者是如何获取消息的

废话不多说，我们开始吧。

首先我们了解一个概念，什么是消费者组。

消费者组你就可以把它理解为，给一组消费者起一个名字。

假设我们有一个订单Topic名字是OrderTopic，然后库存系统和积分系统都要消费这个Topic中的数据，我们分别给库存系统和积分系统起一个消费组名字：stock_consumer_group、score_consumer_group。

设置消费者组名字是在代码中实现的，如下：

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("stock_consumer_group");

比如我们的库存系统提供了2台机器，每台机器上的消费者组名字都是stock_consumer_group，那么这2台机器就是一个消费者组。

大体结构如上图所示，那么当订单系统发送消息到OrderTopic中后，库存系统和积分系统是如何进行消费的呢？

默认情况下，这条消息发送到Broker后，库存系统和积分系统都会拉取这条消息，而且库存系统的两台机器中只有一台会消费到这条消息，积分系统也一样。

这就是消费组的概念，不同的系统设置不同的消费组，如果不同的消费组订阅了同一个Topic，那么对于Topic中的一条消息，每个消费组都会获取到这条消息。

接下来我们思考一个问题，对于消费者组而言，当它获取到一条消息后，假设消费者组内有多台机器，那么到底是只有一台机器获取到消息，还是所有机器都获取到消息呢？

这其实是消费的两种模式，集群模式和广播模式。

默认情况下我们都是使用的集群模式，也就是说消费者组收到消息后，只有其中的一台机器会接收到消息。

我们可以手动指定为广播模式。

consumer.setMessageModel(MessageModel.BROADCASTING)

指定为广播模式后，消费者组内的每台机器都会收到这条消息。

具体要根据业务场景选择消费模式。

接着我们想一下，对于一个Topic下的多个MessageQueue，消费者组中的多台机器是如何消费的呢？

这部分内容底层实现是很复杂的，我们可以简单的理解为它会均匀的将多个MessageQueue分配给消费者组中的多台机器消费。

举个例子，假如我们的OrderTopic有四个MessageQueue，这4个MessageQueue分布在两台MasterBroker上，每个MasterBroker上有两个MessageQueue。

然后库存系统作为一个消费者组有两台机器，那么最好的分配方式就是每台消费者机器负责两个MessageQueue，这样就实现了机器的负载消费，示意图如下：

所以我们可以大致的认为，一个Topic中的多个MessageQueue会被均匀的分布给一个消费者组中的多台机器进行消费，这里要注意一点，一个MessageQueue只能被一台消费者机器消费，但是一台消费者机器可以同时负责处理多个MessageQueue。

那么当消费者组中的机器数量发生变化时，是怎么处理的。

机器数量发生变化一般就两种情况，一种是有机器宕机了，另一种是增加机器进行集群扩容了。

其实这种情况下是会进行rebalance环节的，也就是会重新分配每个消费者机器要处理的MessageQueue。

不知道小伙伴们还记不记得，在之前的文章RocketMQ的发送模式和消费模式中，我们已经用代码说明了消费者的两种消费模式：Push和Pull，当时只提供了Push消费的代码，而没有提供Pull消费的代码。

其实这两种模式本质上是一样的，都是消费者主动发出请求到Broker上拉取消息。

Push模式的底层也是通过消费者主动拉取的方式来实现的，只不过它的名字叫Push而已，意思是Broker尽可能实时的推送消息给消费者。

我们一般在使用RocketMQ的时候，消费模式基本都是使用的Push模式，因为Pull模式真的使用起来代码特别复杂，而且Push模式的底层还是Pull模式，只是对时效性有了更好的支持。

Push模式大体实现思路是这样的：当消费者发送请求到Broker拉取消息的时候，如果有新的消息可以消费，会立马返回消息到消费者进行消费，消费后会接着发送请求到Broker拉取消息。

也就说Push模式下，处理完一批消息后会理解再发送请求给Broker拉取下一批消息，所以时效性更好，看起来就像是Broker在实时推送消息。

当请求发送到Broker发现没有需要消费的消息时，就会让请求线程挂起，默认挂起15秒，然后会有另一个后台线程每隔一段时间判断一下是否有新消息需要消费，一旦发现了新的消息，就会去唤醒挂起的线程，将消息返回给消费者进行消费，然后消费完毕再次发送请求拉取消息。

这一部分的源码实现是很复杂的，我们只要了解它的核心思路就可以了。就算是Push模式，本质上也是对Pull模式的一种封装。

接下来我们来聊聊Broker是如何读取消息返回给消费者的。之前的文章深入研究Broker是如何持久化的中我们已经知道了Broker是如何持久化消息的，小伙伴们可以复习一下。

那么当消费者发送请求到Broker中拉取消息时，假设是第一次拉取，就会从MessageQueue中的第一条消息开始拉取。

如何定位到第一条消息的位置呢，首先Broker会找到MessageQueue对应的ConsumerQueue，从里面找到这条消息的offset，然后通过offset去CommitLog中读取消息数据，把消息返回给消费者。

当消费者消费完这条消息后，会提交一个消费的进度给Broker，Broker会记录下一个ConsumerOffset来标记我们的消费进度。

下次消费者再去这个MessageQueue中拉取消息时，就会从记录的消费位置继续拉取消息，而不用从头获取了。

好了，到这里本篇文章就结束了。