RocketMQ你不得不了解的 Rebalance机制源码分析

以你之姓@ 2023-09-27 23:44 101阅读 0赞

### RocketMQ版本 ###

*  version: 5.1.0

### RocketMQ中consumer消费模型 ###

在了解RocketMQ的Rebalance机制之前，我们必须先简单了解下rocketmq的消费模型

我们知道在我们创建topic的时候需要指定一个参数就是读队列数

![format_png][]

这里假设我们的topic是`xiaozoujishu-topic`，我们的读队列数 是4个，我们同一`gid`下的集群消费模式的消费者有两个，那么我们消费者是如何消费消息的呢 首先需要明确的是:

1.  这里我们的消费模式是集群消费
2.  queue的负载均衡算法是使用默认的`AllocateMessageQueueAveragely`（平均分配） 假设我们项目刚开始只有一个消费者，那么我们的消费队列分配就如下:

![format_png 1][]

四个队列分配给一个消费者

此时如果我们再启动一个消费者，那么这时候就会进行`Rebalance`，然后此时我们的队列分配就变成如下:

![format_png 2][]

所以通过上面的队列分配我就知道`Rebalance`是个啥了，我们下面对`Rebalance`进行一些定义

### RocketMQ的Rebalance是什么 ###

Rebalance(重新平衡)机制指的是：将一个Topic下的多个队列(queue)，在同一个消费者组(consumer group)(gid)下的多个消费者实例(consumer instance)之间进行重新分配

### Rebalance的目的 ###

从上面可以看出`Rebalance`的本意是把一个topic的queue分配给合适的consumer，本意其实是为了提升**消息的并行处理能力**

但是`Rebalance`也带来了一些危害,后面我们会重点分析下

### `Rebalance`的触发原因 ###

我们这里先说结论

1.  订阅Topic的队列数量变化
2.  消费者组信息变化

这里是最深层的原因，就是`topic`的队列数量、`消费组信息` 实际我们可以将这些归结为`Rebalance`的元数据，这些元数据的变更，就会引起clinet的`Rebalance`

> 注意RocketMQ的`Rebalance`是发生在`client`

这些元数据都在管`broker`管理 核心就是这三个类

*  TopicConfigManager
 *  SubscriptionGroupManager
 *  ConsumerManager

只要这个三个类的信息有变化，`client`就会进行`Rebalance`。 下面我们可以具体说下什么情况下会让这三个类变化

#### 订阅Topic的队列数量变化 ####

什么情况下订阅Topic的队列数量会变化呢？

1.  broker扩容
2.  broker缩容
3.  broker宕机(本质也是类似缩容)

#### 消费者组信息变化 ####

什么时候消费者组信息会变化呢？

核心就是consumer的上下线，具体细分又可以分为如下原因：

1.  服务日常滚动升级
2.  服务扩容
3.  服务订阅消息发生变化

### 源码分析 ###

上面大致介绍了`Rebalance`的触发原因，现在我们结合源码来具体分析下

我们就从`consumer`的启动开始分析吧

这里我们以最简单的demo为例

java复制代码    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(CONSUMER_GROUP);
        consumer.setNamesrvAddr(DEFAULT_NAMESRVADDR);
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.subscribe(TOPIC, "*");
        consumer.registerMessageListener((MessageListenerConcurrently) (msg, context) -> {
                System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msg);
    //            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            });
            consumer.start();
    
            System.out.printf("Consumer Started.%n");

这里我们直接注意到 this.mQClientFactory.sendHeartbeatToAllBrokerWithLock(); 这个方法，看名字就知道是client向所有的broker发送心跳

![format_png 3][]

我们进入到sendHeartbeatToAllBrokerWithLock方法看看

java复制代码    private void sendHeartbeatToAllBroker() {
            final HeartbeatData heartbeatData = this.prepareHeartbeatData();
            final boolean producerEmpty = heartbeatData.getProducerDataSet().isEmpty();
            final boolean consumerEmpty = heartbeatData.getConsumerDataSet().isEmpty();
            if (producerEmpty && consumerEmpty) {
                log.warn("sending heartbeat, but no consumer and no producer. [{}]", this.clientId);
                return;
            }
    
            if (this.brokerAddrTable.isEmpty()) {
                return;
            }
            long times = this.sendHeartbeatTimesTotal.getAndIncrement();
            for (Entry<String, HashMap<Long, String>> brokerClusterInfo : this.brokerAddrTable.entrySet()) {
                String brokerName = brokerClusterInfo.getKey();
                HashMap<Long, String> oneTable = brokerClusterInfo.getValue();
                if (oneTable == null) {
                    continue;
                }
                for (Entry<Long, String> singleBrokerInstance : oneTable.entrySet()) {
                    Long id = singleBrokerInstance.getKey();
                    String addr = singleBrokerInstance.getValue();
                    if (addr == null) {
                        continue;
                    }
                    if (consumerEmpty && MixAll.MASTER_ID != id) {
                        continue;
                    }
    
                    try {
                        int version = this.mQClientAPIImpl.sendHeartbeat(addr, heartbeatData, clientConfig.getMqClientApiTimeout());
                        if (!this.brokerVersionTable.containsKey(brokerName)) {
                            this.brokerVersionTable.put(brokerName, new HashMap<>(4));
                        }
                        this.brokerVersionTable.get(brokerName).put(addr, version);
                        if (times % 20 == 0) {
                            log.info("send heart beat to broker[{} {} {}] success", brokerName, id, addr);
                            log.info(heartbeatData.toString());
                        }
                    } catch (Exception e) {
                        if (this.isBrokerInNameServer(addr)) {
                            log.warn("send heart beat to broker[{} {} {}] failed", brokerName, id, addr, e);
                        } else {
                            log.warn("send heart beat to broker[{} {} {}] exception, because the broker not up, forget it", brokerName,
                                    id, addr, e);
                        }
                    }
                }
            }
        }

这段代码主要是通过this.brokerAddrTable.entrySet()获取到所有的master broker地址，然后进行心跳发送

具体的心跳发送代码实际是在下面代码中进行的

java复制代码                    int version = this.mQClientAPIImpl.sendHeartbeat(addr, heartbeatData, clientConfig.getMqClientApiTimeout());

我们进入到该方法会发现和我们之前分析的一样，就是发送一个请求到broker，请求码是RequestCode.HEART\_BEAT

![format_png 4][]

我们看看`RequestCode.HEART_BEAT`的调用找到\`broker的处理逻辑

很快我们通过方法名就能定位到处理client的请求的方法是`ClientManageProcessor`类的`processRequest`

![format_png 5][]

我们具体进去看看这个方法

![format_png 6][]

可以看到具体的逻辑被封装在`return this.heartBeat(ctx, request);`这个方法中，所以我们需要再进去看看

进去这个方法我们能看到一个比较核心的方法 `registerConsumer`

![format_png 7][]

很明显这个方法就是注册consumer的方法

![format_png 8][]

这个方法里面和`Rebalance`相关比较核心的方法就是这三个

1.  `consumerGroupInfo.updateChannel(clientChannelInfo, consumeType, messageModel, consumeFromWhere);`

这里我们可以看看`clientChannelInfo`里面是个啥玩意

![format_png 9][]

具体深入到`updateChannel`方法里面就是判断是否为新的client，是就更新`channelInfoTable`

![format_png 10][]

2.updateSubscription

这个方法就是判断订阅关系是否发生了变化并更新订阅关系

![format_png 11][]

1.  `callConsumerIdsChangeListener`

*  `callConsumerIdsChangeListener(ConsumerGroupEvent.CHANGE, group, consumerGroupInfo.getAllChannel());` 这个方法就是通知client进行`Rebalance`，具体的实现是参考了类似事件订阅的方式去实现的，这里是发送了一个`CHANGE`事件

![format_png 12][]

这里我们可以简单看看事件定义的类型有哪些

![format_png 13][]

我们直接看看具体的事件处理类

![format_png 14][]

可以看到实现类有多个，我们直接看`broker`模块的`DefaultConsumerIdsChangeListener`类即可

![format_png 15][]

可以看到这里是给该group所有的client发送`Rebalance`消息

![format_png 16][]

具体的消息状态码是 `RequestCode.NOTIFY_CONSUMER_IDS_CHANGED`

![format_png 17][]

#### client Rebalance ####

通过上面我们大致找到了整个通信过程，但是实际的`Rebalance`是发生在client，所以我们还是需要继续回到`client`的代码

我们通过状态码`RequestCode.NOTIFY_CONSUMER_IDS_CHANGED` 找到`client`的处理类`ClientRemotingProcessor`

![format_png 18][]

实际处理方法就是

java复制代码this.mqClientFactory.rebalanceImmediately();

我们进入这个方法看看这里最终就是唤醒阻塞的Rebalance线程

![format_png 19][]

所以实际的方法调用还是在`RebalanceService`的 `run`方法

![format_png 20][]

最终还是调用的是`MQConsumerInner`接口中的`doRebalance`方法

这里有个细节，为什么不是直接调用一个静态方法，要搞这么多花里胡哨的唤醒线程操作？

原来是`cleint`也会定时去`Rebalance` 默认是20s一次，可以配置

![format_png 21][]

可以通过参数`rocketmq.client.rebalance.waitInterval`去配置

那么为什么`client`还要自己去循环`Rebalance`

原来这里是防止因为网络等其他原因丢失了`broker`的请求，后续网络回复了，也能进行进行`Rebalance`

下面我们继续看看`Rebalance`的实现细节

这里我们以常用的`DefaultMQPushConsumerImpl`为例

实际这里最终调用的还是抽象类`RebalanceImpl`的`doRebalance`方法

![format_png 22][]

可以看到这里的`Rebalance`是按照`topic`的维度

我们先理解订阅单个`topic`的原理

![format_png 23][]

这里的就是先对`topic`的queue排序，然后对`consumer`排序, 然后调用`AllocateMessageQueueStrategy`的allocate方法 这里我们暂时只分析默认的平均分配算法(AllocateMessageQueueAveragely)，也就是我们最先说的分配算法。其他算法可以详细分析

这里的分配方式就是我们前面画图的，比如4个queue，2个consumer，那么就是每个consumer2个queue。  
简单举例就是我们的 queue有q1、q2、q3、q4  
consumer有 c1、c2

那么就是 c1:q1、q2 c2:q2、q3

> 需要注意的是如果consumer大于queue数量，多出的consumer就不会被分配到queue

### client什么时候触发`Rebalance` ###

上面分析了这么多原理，这里我们总结下`client`什么时候会触发`Rebalance`

1.  consumer启动时会向所有master broker发送心跳，然后`broker`发送信息通知所有`consumer`触发`Rebalance`
2.  启动完成后consumer会周期的触发`Rebalance`，防止因为网络等问题丢失`broker`的通知而没有`Rebalance`
3.  当consumer停止时，也会通过之前分析的事件机制，触发注销`comsuer`事件然后通知所有的`comsuer`触发`Rebalance`

### 总结 ###

这里我们详细介绍了`client`是如何触发`Rebalance`的，以及触发`Rebalance`的时机，也介绍了`Rebalance`的好处。 实际还有很多细节我们限于篇幅暂未分析。 后面我们会继续分析`Rebalance`的坏处和一些详细的`Rebalance`算法

[format_png]: https://img-blog.csdnimg.cn/img_convert/b2c66f845d16c630a73ec2aa8f726d04.webp?x-oss-process=image/format,png
[format_png 1]: https://img-blog.csdnimg.cn/img_convert/f7b00e60a60debda08b480fb92da54de.webp?x-oss-process=image/format,png
[format_png 2]: https://img-blog.csdnimg.cn/img_convert/a6ab0a69cebd96c72c550225bfea12f3.webp?x-oss-process=image/format,png
[format_png 3]: https://img-blog.csdnimg.cn/img_convert/de0f310004ac91f3a5e350349b3cc9e9.webp?x-oss-process=image/format,png
[format_png 4]: https://img-blog.csdnimg.cn/img_convert/50673887506b1ef1c43601cd11869090.webp?x-oss-process=image/format,png
[format_png 5]: https://img-blog.csdnimg.cn/img_convert/968930e5567b96ac4e4cea13e893893e.webp?x-oss-process=image/format,png
[format_png 6]: https://img-blog.csdnimg.cn/img_convert/2010d0145bd0c7df45ec22f868509578.webp?x-oss-process=image/format,png
[format_png 7]: https://img-blog.csdnimg.cn/img_convert/c70bf49e947917dfc4ca34533465ae4b.webp?x-oss-process=image/format,png
[format_png 8]: https://img-blog.csdnimg.cn/img_convert/479e55e9e80ec5f2ce203fc62c3bef26.webp?x-oss-process=image/format,png
[format_png 9]: https://img-blog.csdnimg.cn/img_convert/106a1675a2f9487d06a6d463d1a1ebcf.webp?x-oss-process=image/format,png
[format_png 10]: https://img-blog.csdnimg.cn/img_convert/33aa50677dba76a9ab83c7c29068a835.webp?x-oss-process=image/format,png
[format_png 11]: https://img-blog.csdnimg.cn/img_convert/6791a573239f71ab6f1f8f3736a8570b.webp?x-oss-process=image/format,png
[format_png 12]: https://img-blog.csdnimg.cn/img_convert/89881ca8135b0f788f73d07b1a7da28c.webp?x-oss-process=image/format,png
[format_png 13]: https://img-blog.csdnimg.cn/img_convert/886bbec9677adff512e20e0172dcb06a.webp?x-oss-process=image/format,png
[format_png 14]: https://img-blog.csdnimg.cn/img_convert/b458cbdf960eefc63ae10b7abc597e4e.webp?x-oss-process=image/format,png
[format_png 15]: https://img-blog.csdnimg.cn/img_convert/10da5962afe28d9c2cb8685f7b3eca03.webp?x-oss-process=image/format,png
[format_png 16]: https://img-blog.csdnimg.cn/img_convert/c71380d88747db36a553a3dec993f883.webp?x-oss-process=image/format,png
[format_png 17]: https://img-blog.csdnimg.cn/img_convert/febc81fdd5e6ec21266e9e9c312fc1dc.webp?x-oss-process=image/format,png
[format_png 18]: https://img-blog.csdnimg.cn/img_convert/2f6d0766d0817fd6de32cb96363bb880.webp?x-oss-process=image/format,png
[format_png 19]: https://img-blog.csdnimg.cn/img_convert/0734867bf2fabc3bb17d4e3d0676f8c3.webp?x-oss-process=image/format,png
[format_png 20]: https://img-blog.csdnimg.cn/img_convert/39d76c409a1b4fd90ee84513dde0a26d.webp?x-oss-process=image/format,png
[format_png 21]: https://img-blog.csdnimg.cn/img_convert/e0169a45dd7082789a2802fe5dc27bfa.webp?x-oss-process=image/format,png
[format_png 22]: https://img-blog.csdnimg.cn/img_convert/32244e853156439d050c13c00ab59857.webp?x-oss-process=image/format,png
[format_png 23]: https://img-blog.csdnimg.cn/img_convert/fd51420ff4f971a7c260f2b5b5f92b75.webp?x-oss-process=image/format,png