本文深入的分析了RocketMQ的Rebalance机制，主要包括以下内容：
Rebalance(再均衡)机制指的是：将一个Topic下的多个队列(或称之为分区)，在同一个消费者组(consumer group)下的多个消费者实例(consumer instance)之间进行重新分配。
Rebalance机制本意是为了提升消息的并行处理能力。例如，一个Topic下5个队列，在只有1个消费者的情况下，那么这个消费者将负责处理这5个队列的消息。如果此时我们增加一个消费者，那么可以给其中一个消费者分配2个队列，给另一个分配3个队列，从而提升消息的并行处理能力。如下图：
但是Rebalance机制也存在明显的限制与危害。
Rebalance限制：
由于一个队列最多分配给一个消费者，因此当某个消费者组下的消费者实例数量大于队列的数量时，多余的消费者实例将分配不到任何队列。
Rebalance危害：
除了以上限制，更加严重的是，在发生Rebalance时，存在着一些危害，如下所述：
重复消费：Consumer 2 在消费分配给自己的2个队列时，必须接着从Consumer 1之前已经消费到的offset继续开始消费。然而默认情况下，offset是异步提交的，如consumer 1当前消费到offset为10，但是异步提交给broker的offset为8；那么如果consumer 2从8的offset开始消费，那么就会有2条消息重复。也就是说，Consumer 2 并不会等待Consumer1提交完offset后，再进行Rebalance，因此提交间隔越长，可能造成的重复消费就越多。
消费突刺：由于rebalance可能导致重复消费，如果需要重复消费的消息过多；或者因为rebalance暂停时间过长，导致积压了部分消息。那么都有可能导致在rebalance结束之后瞬间可能需要消费很多消息。
基于Rebalance可能会给业务造成的负面影响，我们有必要对其内部原理进行深入剖析，以便于问题排查。我们将从Broker端和Consumer端两个角度来进行说明：
Broker端主要负责Rebalance元数据维护，以及通知机制，在整个消费者组Rebalance过程中扮演协调者的作用；
而Consumer端分析，主要聚焦于单个Consumer的Rebalance流程。
从本质上来说，触发Rebalance的根本因素无非是两个：1 ) 订阅Topic的队列数量变化 2)消费者组信息变化。导致二者发生变化的典型场景如下所示：
1 队列信息变化 |
典型场景：
|
2 消费者组信息变化 |
典型场景：
|
在这里，笔者将队列信息和消费者组信息称之为Rebalance元数据，Broker负责维护这些元数据，并在二者信息发生变化时，以某种通知机制告诉消费者组下所有实例，需要进行Rebalance。从这个角度来说，Broker在Rebalance过程中，是一个协调者的角色。
在Broker内部，通过元数据管理器维护了Rebalance元数据信息，如下图所示：
这些管理器，内部实现都是一个Map。其中：
队列信息：由TopicConfigManager维护。Map 的key是Topic名称，Value是TopicConfig。Broker通过实时的或者周期性的上报自己的Topic配置信息给NameServer，在NameServer组装成Topic的完整路由信息。消费者定时向NameServer定时拉取最新路由信息，以实现间接通知，当发现队列信息变化，触发Rebalance。
消费者组信息：由ConsumerManager、ConsumerOffsetManager、SubscriptionGroupManager三者共同维护。ConsumerManager维护了消费者组订阅信息，以及消费者组下当前的消费者实例信息，当消费者组的订阅信息或者实例发生变化，Broker都会主动给所有消费者实例发送通知，触发Rebalance。而在Rebalance时，消费者需要从ConsumerOffsetManager查询应该从那个位置继续开始消费。SubscriptionGroupManager主要是维护消费者组的一些附加信息，方便运维。
2.1 队列信息变化
队列信息通过Broker内的TopicConfigManager来维护，每个Broker都会将自己的信息上报给NameServer，由NameServer组装成完整的Topic路由信息。
通常情况下，一个Topic下的队列数量不会频繁的变化，但是如果遇到，Topic队列数量扩/缩容，、broker日常运维时的停止/启动或者broker异常宕机，也有可能导致队列数量发生变化。
这里我们重点讲一下为什么broker异常停止/宕机会导致数量变化。一些读者可能会认为创建Topic时，已经明确指定了队列的数量，那么之后不论怎样，队列的数量信息都不会发生变化，这是一种典型误解。
下图展示了一个RocketMQ集群双主部署模式下，某个broker宕机后，Topic路由信息的变化。
可以看到，在宕机前，主题TopicX下队列分布在broker-a和broker-b两个broker上，每个broker上各有8个队列。当broker-a宕机后，其路由信息会被移除，此时我们就只能看到TopicX在broker-b上的路由信息。
因此，在RocketMQ中，Topic的路由信息实际上是动态变化的。不论是停止/启动/扩容导致的所有变化最终都会上报给NameServer。客户端可以给NameServer发送GET_ROUTEINTO_BY_TOPIC请求，来获得某个Topic的完整路由信息。如果发现队列信息发生变化，则触发Reabalance。
2.2 消费者组信息变化
Rebalance的另外一个条件：消费者组信息，Broker端通过以下三个组件共同维护：
ConsumerManager：维护消费者实例信息和订阅信息
ConsumerOffsetManager：维护offset进度信息
SubscriptionGroupManager：运维相关操作信息维护
ConsumerManager是最重要的一个消费者组元数据管理器，其维护了某个消费者组的订阅信息，以及所有消费者实例的详细信息，并在发生变化时提供通知机制。
数据添加：
客户端通过发送HEART_BEAT请求给Broker，将自己添加到ConsumerManager中维护的某个消费者组中。需要注意的是，每个Consumer都会向所有的Broker进行心跳，因此每个Broker都维护了所有消费者的信息。
数据删除：
客户端正常停止时，发送UNREGISTER_CLIENT请求，将自己从ConsumerManager移除；此外在发生网络异常时，Broker也会主动将消费者从ConsumerManager中移除。
数据查询：
消费者可以向任意一个Broker发送GET_CONSUMER_LIST_BY_GROUP请求，来获得一个消费者组下的所有消费者实例信息。
我们可以通过mqadmin命令行工具的consumerConnection子命令，来查看某个消费者的信息，如：
输出主要分为2个部分：
消费者组实例信息：展示了groupA下当前有2个消费者，以及对应的详细信息，包括：消费者id，消费者ip/port，消费者语言，消费者版本。
消费者组订阅信息：包括订阅的Topic，过滤条件，消费模式，以及从什么位置开始消费等。
这二者不论哪个信息发生变化，Broker都会主动通知这个消费者组下的所有实例进行Rebalance。在ConsumerManager的registerConsumer方法中，我们可以看到这个通知机制。如以下源码片段红色框中所示：
consumerIdsChangeListener在处理消费者组信息变更事件时，会给每个消费者实例都发送一个通知，各个消费者实例在收到通知后触发Rebalance，如下图所示：
敏锐读者注意到了，Broker是通知每个消费者各自Rebalance，即每个消费者自己给自己重新分配队列，而不是Broker将分配好的结果告知Consumer。
从这个角度，RocketMQ与Kafka Rebalance机制类似，二者Rebalance分配都是在客户端进行，不同的是：
此时，我们需要思考2个问题：
问题1：每个消费者自己给自己分配，如何避免脑裂的问题呢？
因为每个消费者都不知道其他消费者分配的结果，会不会出现一个队列分配给了多个消费者，或者有的队列分配给了多个消费者。
问题2：如果某个消费者没有收到Rebalance通知怎么办？
每个消费者都会定时触发Rebalance，以避免Rebalance通知丢失。
2.2.2 ConsumerOffsetManager
事实上，通过ConsumerManager已经可以获得Rebalance时需要的消费者所有必要信息。但是还有一点，Rebalance时，如果某个队列重新分配给了某个消费者，那么必须接着从上一个消费者的位置继续开始消费，这就是ConsumerOffsetManager的作用。
消费者发送UPDATE_CONSUMER_OFFSET请求给Broker，来更新消费者组对于某个Topic的消费进度。发送QUERY_CONSUMER_OFFSET请求，来查询消费进度。
通过mqadmin命令行工具的consumerProgress子命令，来可以看到Topic每个队列的消费进度，如：
2.2.3 SubscriptionGroupManager
订阅组配置管理器，内部针对每个消费者组维护一个SubscriptionGroupConfig。主要是为了针对消费者组进行一些运维操作，这里不做过多介绍，感兴趣的读者自行查阅源码。
前面分析Broker在Rebalance过程中起的是协调通知的作用，可以帮忙我们从整体对Rebalance有个初步的认知。但是Rebalance的细节，却是在Consumer端完成的。
在本节中，我们将着重讨论单个consumer的Rebalance流程。
需要说明的是，RocketMQ的consumer分配pull和push两种模式，二者的工作逻辑并不相同。这里主要以push模式的默认实现类DefaultMQPushConsumer为例进行讲解。
在前文，我们提到Broker会主动通知消费者进行Rebalance，但是从消费者的角度来看，整个生命过程的各个阶段，都有可能触发Rebalance，而不仅仅是收到通知后才进行Rebalance。
具体来说，Consumer在启动/运行时/停止时，都有可能触发Rebalance，如下图所示：
下面通过源码分析，分别讲解启动时/运行时/停止时是如何触发Rebalance的。
DefaultMQPushConsumerImpl的start方法显示了一个消费者的启动流程，如下图所示：
可以看到Consumer启动主要分为5个步骤，其中步骤2、4、5是我们分析的重点。：
步骤2 ：更新订阅的topic路由信息
上述代码步骤2，调用updateTopicSubscribeInfoWhenSubscriptionChanged()方法，从NameServer更新topic路由信息，由于一个消费者可以订阅多个topic，因此这个Topic都需要更新，如下：
通过这一步，当前Consumer就拿到了Topic下所有队列信息，具备了Rebalance的第一个条件。
步骤4 向broker发送心跳信息
在上述启动流程中的第4步，调用sendHeartbeatToAllBrokerWithLock方法，给每个Broker都发送一个心跳请求。
this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
当Broker收到心跳请求后，将这个消费者注册到ConsumerManager中，前文提到，当Consumer数量变化时，Broker会主动通知其他消费者进行Rebalance。
而心跳的数据，这些数据是在MQClientInstance类的prepareHeartbeatData方法来准备的。我们在前文通过mqadmin命令行工具的consumerConnection 自命令查看到的消费者订阅信息，在这里都出现了，如下图红色框所示：
提
示：
可以看到心跳数据HeartbeatData中，既包含Consumer信息，也包含Producer信息
(这里进行了省略)
。
步骤5：立即触发一次Rebalance
消费者启动流程的最后一步是调用以下方法立即触发一次rebalance：
this.mQClientFactory.rebalanceImmediately();
这个方法内部实际上，是通过唤醒一个RebalanceService，来触发Rebalance：
public void rebalanceImmediately() { this.rebalanceService.wakeup(); }
这里我们并不着急分析RebalanceService的内部具体实现，因为所有的Rebalance触发都是以这个类为入口，我们将在讲解完运行时/停止时的Rebalance触发时机后，统一进行说明。
消费者在运行时，通过两种机制来触发Rebalance：
下面分别进行说明：
1 监听broker 消费者数量变化通知，触发rebalance
RocketMQ支持双向通信机制，在客户端通过ClientRemotingProcessor的processRequest方法来处理Broker发起的通知请求，如下：
目前，我们关注的是，消费者数量变化时，Broker给客户端的通知，也就是上图中红色框的内容。在收到通知后，其调用notifyConsumerIdsChanged进行处理，这个方法内部会立即触发Rebalance。
可以看到这里是调用mqClientFactory的rebalanceImmediately方法触发Rebalance，而前面讲解消费者启动时是通过RebalanceService触发，事实上，后者RebalanceService内部也是通过mqClientFactory进行触发Rebalance。
2 周期性触发rebalance，避免Rebalance通知丢失
为了避免Broker的Rebalance通知丢失问题，客户端还会通过RebalanceService定时的触发Rebalance，默认间隔是20秒，如下图：
3.3 停止时触发
最后，消费者在正常停止时，需要调用shutdown方法，这个方法的工作逻辑如下所示：
在停止时，会首先通过第2步持久化offset，前文提到过默认情况下，offset是异步提交的，为了避免重复消费，因此在关闭时，必须要对尚未提交的offset进行持久化。
其实就是发送更新offset请求(
UPDATE_CONSUMER_OFFSET)
给Broker，
Broker对应更新ConsumerOffsetManager中的记
录。
这样当队列分配给其他消费者时，就可以从这个位置继续开始消费。
接着第3步调用unregisterConsumer方法，向所有broker发送UNREGISTER_CLIENT命令，取消注册Consumer。broker接收到这个命令后，将consumer从ConsumerManager中移除，然后通知这个消费者下的其他Consumer进行Rebalance。
至此，我们已经讲解完了Consumer启动时/运行时/停止时，所有可能的Rebalance触发时机，在下一小节，将介绍消费者Rebalance具体步骤。
4 Consumer Rebalance流程
前面花了大量的篇幅，讲解了Rebalance元数据维护，Broker通知机制，以及Consumer的Rebalance触发时机，目的是让读者有一个更高层面的认知，而不是直接分析单个Consumer Rebalance的具体步骤，避免一叶障目不见泰山。
不同的触发机制最终底层都调用了MQClientInstance的doRebalance方法，而在这个方法的源码中，并没有区分哪个消费者组需要进行Rebalance，只要任意一个消费者组需要Rebalance，这台机器上启动的所有其他消费者，也都要进行Rebalance。相关源码如下所示：
MQClientInstance#doRebalance
上述代码逐一迭代当前机器启动的所有消费者(MQConsumerInner)，并调用其doRebalance方法进行触发Rebalance。
MQConsumerInner有push模式和pull模式两种实现，分别是：
其会根据消费者指定的消息监听器是有序还是无序进行判定Rebalance过程中是否需要对有序消费进行特殊处理。
参见：
DefaultMQPushConsumerImpl#doRebalance。
总是认为是无序的，因为写死了为false。
参见：
DefaultMQPullConsumerImpl#doRebalance
我们看到，不管是push还是pull模式的Consumer实现，内部都是调用RebalanceImpl的doRebalance方法进行触发，将是否有序作为一个参数传入。
在这个方法内部，如果一个消费者订阅了多个Topic，会迭代每个Topic维度逐一触发Rebalance。相关源码如下所示：
RebalanceImpl#doRebalance
RocketMQ按照Topic维度进行Rebalance，会导致一个很严重的结果：如果一个消费者组订阅多个Topic，可能会出现分配不均，部分消费者处于空闲状态。
举例来说：某个消费者组group_X下有4个消费者实例，分别部署在192.168.0.[1-4] 四台机器上；订阅了两个主题：TopicX和TopicY。如下图：
其中：001~004表示的就是这4个消费者的信息，而订阅信息显示了订阅TopicX和TopicY。
假设TopicX、TopicY各有2个队列，因此总共有4个队列；而刚好又有4个消费者，我们的期望是每个消费者分配一个队列。然后实际分配情况如下图所示：
通过观察Client IP列，我们看到192.168.0.1、192.168.0.2各出现2两次，也就是分配到了两个队列，另外2个IP(192.168.0.3、192.168.0.4)并没有出现，表示没有分配到任何队列。
之所以出现分配不均，就是因为按照Topic维度进行Rebalance，因此这里TopicX和TopicY会各Rebalance一次。且每次Rebalance时都对消费者组下的实例进行排序，所以TopicX和TopicY各自的两个队列，都分配给消费者组中的前两个消费者了。
由于订阅多个Topic时可能会出现分配不均，这是在RocketMQ中我们为什么不建议同一个消费者组订阅多个Topic的重要原因。在这一点上，Kafka与不RocketMQ同，其是将所有Topic下的所有队列合并在一起，进行Rebalance，因此相对会更加平均。
单个Topic的Rebalance流程，是在RebalanceImpl类的rebalanceByTopic方法中进行的，整体上可以分为3大步骤：
1 获得Rebalance元数据信息
2 进行队列分配
3 分配结果处理
如以下源码截图红色框中所示：
下面对每一个步骤进行详细说明。
4.2.1 获得Rebalance元数据
消费者在Rebalance时需要获得：Topic的队列信息和消费者组实例信息。
对于队列信息：
会从之前的缓存的Topic路由信息中获取；Topic路由信息会定时的进行更新。
对于消费者组实例信息：
前面我们提到过Broker通过ConsumerManager维护了所有的消费者信息，findConsumerIdList方法内部会会发送GET_CONSUMER_LIST_BY_GROUP给请求给任意一个Broker进行获取。
4.2.2 进行队列分配
RocketMQ的分配策略使用AllocateMessageQueueStrategy接口表示，并提供了多种实现：
AllocateMessageQueueAveragely：平均分配，默认
AllocateMessageQueueAveragelyByCircle：循环分配
AllocateMessageQueueConsistentHash：一致性哈希
AllocateMessageQueueByConfig：根据配置进行分配
AllocateMessageQueueByMachineRoom：根据机房
AllocateMachineRoomNearby：就近分配
这里举例来进行说明。
假设某个Topic有10个队列，消费者组有3个实例c1、c2、c3，使用AllocateMessageQueueAveragely分配结果如下图所示：
因为这是一个平均分配策略，在分配时，每个消费者(c1、c2、c3)平均分配3个，此时还多出1个，多出来的队列按顺序分配给消费者队列的头部元素，因此c1多分配1个，最终c1分配了4个队列。
需要注意的是，每个消费者是自己给自己分配，相当于存在多个大脑。那么如何保证分配结果的一致呢？
通过以下两个手段来保证：
对Topic队列，以及消费者各自进行排序
每个消费者需要使用相同的分配策略。
尽管每个消费者是各自给自己分配，但是因为使用的相同的分配策略，定位从队列列表中哪个位置开始给自己分配，给自己分配多少个队列，从而保证最终分配结果的一致。
对于其他分配策略，感兴趣的读者可以自行阅读源码，在实际开发中使用的很少，这里并不介绍。
特别
的，mqadmin工具提供了一个allocateMQ子命令，通过其我们可以预览某个Topic在多个消费者分区是如何分配的，使用方式如下：
sh bin/mqadmin allocateMQ -i ip1,ip2,ip3 -t TopicA -n localhost:9876
这个工具可以将模拟分配的结果进行json格式展示。
处理队列变更
消费者计算出分配给自己的队列结果后，需要与之前进行比较，判断添加了新的队列，或者移除了之前分配的队列，也可能没有变化。
可参考updateProcessQueueTableInRebalance()实现。
其他处理
调用messageQueueChanged方法进行额外后续处理：
对于push模式：
主要是进行一些流控参数的更新。
对于pull模式：
是回调用户自定义的MessageQueueListener。
长按二维码，关注我，加好友，进群交流
往期精彩
