RocketMQ源码解析之消息消费者(rebalance)_this.mqclientfactory.rebalanceimmediately()

原创不易，转载请注明出处

前言

从本文开始我们就进入消息消费者核心机制介绍了，其实消息消费就三大核心步骤，第一就是rebalance，第二就是拉取消息，第三就是消费，对消息消费中出现的异常做一些处理。在阅读本篇文章之前最好阅读一下《RocketMQ源码解析之消息消费者（发送心跳给broker）》这篇文章，这篇主要介绍了消息消费者向namesrv拉取topic路由信息与向broker 进行注册这两个功能，其实这个向broker注册是比较重要的，它会将消息消费者的一写基本信息，订阅信息带给broker ，同时broker发现这个消费者组的成员发生了变化（毕竟是新加入一个消息消费者），这个时候broker 就会发消息通知这个消费者组下面的所有消费者说你们这个消费者组的成员发生了变化，然后需要你们重新rebalance，本文先从rebalance 原理出发，将原理介绍清楚，然后带着原理去源码中找到具体的实现。

1.rebalance原理介绍

这里先介绍下 rebalance 是啥意思，从字面意思上是重新负载均衡，其实在RocketMQ中你也可以这样理解，但是重新负载均衡什么呢？我们拿集群消息的消费举例子，集群消息同一个消费者组只能有一个消费者消费这个消息，这个样子如果你有一个topicA这么主题，分成了8个MessageQueue，然后有消费者组groupA，一开始的时候，就这个消费者组就一个消费者实例，这个时候你一个消费者实例就要消费这8个queue，如下图

这个时候，你有起了一个groupA组的消费者实例2 ，然后这个消费者2 启动的时候会向broker进行注册，然后broker 发现这个groupA组新加了一个消费者实例，这个时候，就会通知这个消费者实例1说你们这个组的消费者实例发生了变化，请立即重新负载均衡一下，消费者实例2发送完注册过程，其实它也是会走这个重新负载的步骤。
这里我们就挑消费者实例2说吧，它首先会从broker 拉取这个组的消费者实例id，发现就2两个，同时获取订阅topic的MessageQueue信息集合，一看是8个，然后就根据负载算法（这里使用的是平均分配的算法），这下好办了，8个queue，然后2个实例，一人4个queue正好均分

这个时候就是上图这个样子了，实例1跟实例2一人4个queue，拿到这个queue之后就要进行处理了，他会将以前别的queue清理掉，就像实例1一样，它以前有queue4567 ，现在没有了，它就要将本地维护的一个ProcessQueue清理掉，这个ProcessQueue现在先不用管他，其实就是一个处理消费的queue，他与你的MessageQueue是一一对应的，然后就像消费者实例2这个样子的，新加了4个queue，他就会遍历这4个queue，获取每个queue 从哪个offset 开始消费（这个是根据你设置的ConsumeFromWhere参数来决定的），然后为每个queue生成一个对应的ProcessQueue，最后封装拉取请求，将这些拉取消息请求信息投递到PullMessageService 这么一个组件中去，由这个组件进行后续的拉取消息动作。
到这我们rebalance原理就介绍完了，简单说就是同一个消费者组的不同消费者怎么分配MessageQueue这么一个事情，接下来我们就来看下源码是什么样子的。

2.源码剖析

我们要从消费者DefaultMQPushConsumerImpl启动时候的 this.mQClientFactory.rebalanceImmediately();开始说起

这里调用了rebalance服务的唤醒，RebalanceService其实是个线程，在MQClientInstance启动的时候会启动这个线程服务
但是一上来就是等待20s，接着 this.mQClientFactory.rebalanceImmediately();唤醒了这个线程，立即执行doRebalance。

可以看到，调用每个消费对象的doRebalance，一个MQClient是可以对应多个DefaultMQPushConsumer ，DefaultMQProducer的

然后DefaultMQPushConsumerImpl找了rebalanceImpl来具体干这个事情，这个consumerOrderly就是看你是不是顺序消费，是通过是编码的MessageListener类型来判断的。

这里就是遍历所有订阅的topic来根据topic重新负载均衡，接下来我们看下rebalanceByTopic这个方法的实现，这个方法里面按照是集群消息还是广播消息做了判断，我们主要看下这个集群消息的rebalance是怎么做的

我这里将rebalance方法大体分了5个步，这里其实就是整个rebalance最最最核心的了，我们一步一步来介绍
第一步，主要是从topic订阅表中获取这个topic的所有MessageQueue，这些MessageQueue其实就是定时从namesrv拉取回来的，没印象的可以看下《RocketMQ源码解析之消息消费者（发送心跳给broker）》这篇文章，然后根据topic 与消费者组去broker 上面获取这个组的所有消费者实例，接下来就是一堆判断了。
第二步，就是排序了，将MessageQueue集合进行排序，将该组下的消费者实例集合进行排序，这个一步非常的重要，我们第三步的将负载均衡算法看完就知道了
第三步，使用负载均衡算法进行重新负载，这里使用默认的AllocateMessageQueueAveragely 平均分配queue的算法。接着看下allocate具体实现。

其实前面还一堆的判断，我们这里就不看了，直接看下算法的实现， 先用 队列的数量% 消费者数量，这个其实就是想看看 这几个消费者能不能均分这几个队列，不能的话还剩几个零头的队列，接着就是计算你自己能分几个，如果能均分的话，你跟别人都一样，如果不能均分的话，就看看你在消费者里面哪个位置（其实这里就体现出第二步排序的重要性了），如果你的位置在零头个数的前头的话，就还能分一个队列，这里的核心就是如果能均分就均分，不能均分的话看看还剩几个队列，比如剩3个把，就让排在前面的三个消费者实例一人一个。
接下来就是判断从哪个位置开始取了，比如说我现在有8个queue，然后3个消费者实例分别是0，1，2，0是自己，然后按照上面的算法，消费者0能分3个queue，消费者1也能分3个queue，消费者2分2个queue，如下图：

接着就是计算从queue集合取的位置了，根据上面的那个公式算出来，消费者0 是从0 开始取，消费者1是从3开始取，消费者2是从6开始取，然后计算出取的范围，取的范围与你分几个queue有很大的关系

接着就是具体的取了，如果是消费者0的话就是从0开始取，取3个，然后消费者1的话从3开始取往后取3个，消费者2的话从6开始去取2个，如下图

仔细体会一下，如果排序的话，会是什么情况，可能会分到重复的queue。这个时候，我自己是消费者0 ，我就可以把queue1，2，3 返回去了。
第四步，更新这个ProcessQueue，这个ProcessQueue与MessageQueue是一一对应的，后面的消息消费就是使用这个ProcessQueue，从名字上面也能看出来，这块内容也比较多，我们一部分部分一步的看。

这里其实就是遍历processQueueTable 这个map，它维护了MessageQueue与ProcessQueue的对应关系，如果新分配的queue里面没有之前那个了，就将之前的那个移除，如果之前的queue过了很长时间没有拉取数据，也就是代码中超过120s没有拉回来数据了，也会将它销毁，看看怎么销毁的，先是设置ProcessQueue的状态为销毁，然后移除没必要的queue，这块就是将这个queue的消费offset上报broker，然后如果是顺序消费的话，向broker释放锁等等

我们移除了那些超时的queue与不存在的queue 之后，就要处理新增的了，遍历queue集合，然后如果之前没有的话，获取一下这个MessageQueue的消费位置，就是从哪开始消费，如果是CONSUME_FROM_LAST_OFFSET 的话，就要从broker 上获取了，获取消费到哪了，如果broker没有这个queue的消费offset信息，就会返回-1，这个时候它就会再从broker获取这个queue最大的offset，也就是从最后面开始消费，如果你配置了CONSUME_FROM_FIRST_OFFSET，也是先去broker上获取消费到哪个offset了，如果没有的话从0开始消费，如果配置了CONSUME_FROM_TIMESTAMP，也是从broker先获取最后消费offset，如果没有话，就去broker上找你这个时间的一个offset。
我这里省掉了不少细节，这块可以自己看下。
接着就是组装ProcessQueue塞到map中了，组装PullRequest，放到list中，分发出去，分发这动作由子类完成。如果有改变就返回true，没有变动就返回false
第五步，如果有变动的话就要通知MessageQueue发生了变化，这块也是由子类完成，我们这里看下RebalancePushImpl 类的实现

这块其实就是计算一些拉取消息的阈值了，如果你配置了topic级别的拉取大小的阈值，然后它会计算出来每个queue的拉取阈值，就是topic的限制/queue的数量，然后设置到DefaultMQPushConsumer的pullThresholdForQueue 参数中。同理，下面消息大小的阈值也是这样子计算的。最后向broker 发送心跳。
好了，我们rebalance基本解析完成了，最后我们再来看下第四步RebalancePushImpl这个子类是怎样实现 PullRequest 的分发动作的

这块其实是循环找DefaultMQPushConsumer 的立即执行PullRequest方法，这一块再往下内容我们介绍消息拉取的时候会详细介绍

总结

本篇主要是介绍了RocketMQ 消息消费者push模式 rebalance的实现原理，rabalance其实就是解决一堆MessageQueue怎么分给一堆消费者的问题，最后又解析了rebalance相应功能的源码实现。