• 聊聊RocketMQ Rebalance负载均衡触发机制浅析

    想必你已经了解RocketMQ的设计哲学是，同一时刻同一个Queue只能被一个Consumer持有，
    但同一个Consumer可以同时消费多个队列，为了订阅、消费模型的高效，Rocket总是希望将Queue分配的足够均匀，
    日常使用时，Consumer的上下线，Queue的动态扩缩容，都可能会破坏分配均衡性，
    故而Rocket提供了一套完整的Rebalance机制针对上述状况。

    触发条件

    1.  总结下来Rebalance一共三个触发条件，两个主动，一个被动。满足任意一个都会触发
    2.  
    3.  1.Consumer启动之时执行start方法主动执行负载均衡逻辑；
    4.  2.定时任务触发；
    5.  3.Broker下发通知告知Client需要进行负载均衡；
    6.  
    7.  今天重新翻阅代码的时候发现
    8.  很巧合三个触发条件或多或少跟DefaultMQPushConsumerImpl.start()都有关系;
    9.

    DefaultMQPushConsumerImpl.start()

    DefaultMQPushConsumerImpl创建实例时，会初始化rebalanceImpl成员变量 private final RebalanceImpl rebalanceImpl = new RebalancePushImpl(this); 此时此刻这个rebalanceImpl对象没有一点作用，因为ta的关键成员属性尚且为null，下文中的start肩负起了赋值重任。
    下面是摘录start()主要代码:

    1. public synchronized void start() throws MQClientException {
    2.     switch (this.serviceState) {
    3.         case CREATE_JUST: 
    4.             /* 检查配置 */
    5.             this.checkConfig();
    6.             /* 构建 Topic 订阅信息——SubscriptionData，并添加至 RebalanceImpl 的订阅信息中 */
    7.             this.copySubscription();
    8.             /* 初始化 MQClientInstance */
    9.             this.mQClientFactory = MQClientManager.getInstance()
    10.                 .getOrCreateMQClientInstance(this.defaultMQPushConsumer, this.rpcHook);
    11.            /**
    12.             * 丰富 rebalanceImpl 对象属性，注意到了吗之前初始化的对象充血了
    13.             * 之前产生的 rebalanceImpl 对象直到此刻才算真正意义上的初始化完成
    14.             * rebalanceImpl就是负载均衡的相关实现
    15.             */ 
    16.             this.rebalanceImpl.setConsumerGroup(this.defaultMQPushConsumer.getConsumerGroup());
    17.             this.rebalanceImpl.setMessageModel(this.defaultMQPushConsumer.getMessageModel());
    18.             this.rebalanceImpl.setAllocateMessageQueueStrategy(this.defaultMQPushConsumer.getAllocateMessageQueueStrategy());
    19.             this.rebalanceImpl.setmQClientFactory(this.mQClientFactory);
    20.  
    21.             /**
    22.              * 向 MQClientInstance 注册消费者，并启动 MQClientInstance
    23.              * 一个 JVM 中的所有消费者、生产者持有同一个 MQClientInstance，MQClientInstance 只会启动一次
    24.              */
    25.             boolean registerOK = mQClientFactory.registerConsumer(this.defaultMQPushConsumer.getConsumerGroup(), this);
    26.             break;  
    27.         case 
    28.             ...;
    29.         default:
    30.             break;  
    31.     }
    32.  
    33.     /* Consumer启动成功，立即向所有Broker发送心跳 */
    34.     this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
    35.  
    36.     /*
    37.      * 注意到了嘛，Consumer上线会立即触发一次负载均衡 
    38.      * 但是这里并不是调用一下负载均衡的实现那么简单，这里其实是唤醒了相关服务线程
    39.      * 下文笔者会着重介绍
    40.      */
    41.     this.mQClientFactory.rebalanceImmediately();
    42. }
    43.

    RebalanceImpl

    见名知意，Consumer负载均衡相关的操作全部都委托给RebalanceImpl对象。 每一个Consumer的对象都持有一个RebalanceImpl实例,每个RebalanceImpl实例也只服务于一个Consumer。
    二者是一个相互持有，循环引用的关系。
    我们来看一下这个对象的关键成员属性:

    1. RebalancePushImpl extends RebalanceImpl { 
    2.     protected final ConcurrentMap<MessageQueue, ProcessQueue> processQueueTable = new ConcurrentHashMap<>(64);
    3.  
    4.     /* ConcurrentMap<topic, Set<MessageQueue>>, Topic与分给自己的MessageQueue信息 */
    5.     protected final ConcurrentMap<String, Set<MessageQueue>> topicSubscribeInfoTable = new ConcurrentHashMap<>();
    7.     /* ConcurrentMap<topic,  SubscriptionData>, Topic与订阅数据 */
    8.     protected final ConcurrentMap<String, SubscriptionData> subscriptionInner = new ConcurrentHashMap<>();
    9.  
    10.     /* 负载算法的具体实现，究竟如何分配就是由这个总指挥决定的 */
    11.     protected AllocateMessageQueueStrategy allocateMessageQueueStrategy;
    12.  
    13.     /* Consumer实例 */
    14.     private final DefaultMQPushConsumerImpl defaultMQPushConsumerImpl;
    15. }
    16.

    其中有一个ProcessQueue对象尤为瞩目，因为我没做任何注释，如果你有阅读Rocket源码应该知道他是 Consumer消费Message过程中极其重要的一环，举足轻重，你可以认为ta是Client端的消息承载者。 因为跟负载时机关系不大所以此处不再赘述。
    负责进行均衡负载的就是doRebalance()，实际上真正执行负载逻辑的是rebalanceByTopic();

    RebalanceByTopic()

    rebalanceByTopic()是负载均衡的最终落脚点，即系统中所有需要负载的调用最后都会走到这里来。
    集群消费模式下的实现 摘录关键代码:

    1. private void rebalanceByTopic(String topic, boolean isOrder) {
    2.     /* 获取到该Topic下的所有队列 */
    3.     Set<MessageQueue> mqSet = this.topicSubscribeInfoTable.get(topic);
    4.     /* 获取该Topic下ConsumerGroup此消费组所有的消费者Id */
    5.     List<String> cidAll = this.mQClientFactory.findConsumerIdList(topic, consumerGroup);
    6.     if (mqSet != null && cidAll != null) {
    7.         List<MessageQueue> mqAll = new ArrayList<>(mqSet);
    8.  
    9.         /* 这两个排序极其关键 */
    10.         Collections.sort(mqAll);
    11.         Collections.sort(cidAll);
    12.  
    13.         /* 负载均衡算法 */
    14.         AllocateMessageQueueStrategy strategy = this.allocateMessageQueueStrategy;
    15.         List<MessageQueue> allocateResult;
    16.         /* 调用具体的算法实现进行队列分配 */
    17.         allocateResult = strategy.allocate(
    18.             this.consumerGroup,
    19.             this.mQClientFactory.getClientId(),
    20.             mqAll,
    21.             cidAll
    22.         );
    23.     }
    24. }
    25.

    看似轻描淡写实则举足轻重的两句代码:
    Collections.sort(mqAll);
    Collections.sort(cidAll);
    这两句代码意义在于该ConsumeGroup下的所有Consumer得到的队列顺序、消费者Id顺序都是一致的。在分配视图一致性得到保证的前提下，分配算法又是相同的，这样一来尽管各个Consumer在负载均衡的时候不进行任何信息交换，但是却可以互不干扰有条不紊的将队列均衡的分配完毕。
    如何分配的具体细节在allocateMessageQueueStrategy中，RocketMQ也默认支持多种分配算法，比较简单，笔者不想赘述。
    负载均衡的实现讲完了，那谁会调用ta，如何调用ta，在什么时机下会调用ta，一个个问题萦绕心头，分析上述问题之前我们总是绕不开RebalanceService对象。

    RebalanceService

    RocketMQ中有一类对象地位超然，特立独行，一个对象主宰一个领域，ta们往往只受操作系统的约束。 似乎操作系统也格外偏爱他们，因为会给他们分配时间片，直接调度他们的运行(其实原因很简单，后面会有答案)。 ta就是传说中的ServiceThread，江湖中人往往称呼ta们为服务线程。

    1. public abstract class ServiceThread implements Runnable {
    2.     protected boolean isDaemon = false;
    3.  
    4.     /* 这能不被钟爱吗，直接持有一个独立线程 */
    5.     private Thread thread;
    6.  
    7.     /* 执行start的时候申请一个线程 */
    8.     public void start() {
    9.         /* 只允许申请一次 */
    10.         if (!started.compareAndSet(false, true)) {
    11.             return;
    12.         }
    13.         stopped = false;
    14.         this.thread = new Thread(this, getServiceName());
    15.         /* 设置为非守护线程 */
    16.         this.thread.setDaemon(isDaemon);
    17.         this.thread.start();
    18.     }
    19. }
    20.

    ServiceThread家族兴旺，除了专门负责负载均衡的RebalanceService，还有一众兄弟姐妹:
    FlushRealTimeService:异步刷盘服务线程
    CommitRealTimeService:异步刷盘服务线程
    GroupCommitService: 同步刷盘服务线程
    ......
    每一个都是Rocket能够稳定运行的背后功臣，上述三个其实就是刷盘相关的服务线程。 (涉及的面太广，如果有人想看，我尝试分析一下)

    有了上面的铺垫，RebalanceService应该很好理解了，单独持有一个线程进行负载均衡，当然ta也不是无休止的一直进行负载处理。

    1. public class RebalanceService extends ServiceThread { 
    2.     /* 负载均衡时间间隔，默认20s，支持配置 */
    3.     private final static long waitInterval =Long.parseLong(
    4.         System.getProperty("rocketmq.client.rebalance.waitInterval", "20000")
    5.     );
    6.  
    7.     private final MQClientInstance mqClientFactory;
    8.  
    9.     public void run() {
    10.         /* 只要该线程未终止就会一直执行 */
    11.         while (!this.isStopped()) {
    12.             /* 喜欢才会放肆，但爱是克制，休息20s */
    13.             this.waitForRunning(waitInterval);
    14.             /* 执行负载均衡 */
    15.             this.mqClientFactory.doRebalance();
    16.         }
    17.     }
    18. }
    19.

    定时任务

    1. 理解了上面的RebalanceService，应该就理解了定时触发的逻辑，只需要定时唤醒服务线程即可
    2. 每个Java服务单点只会启动一个RebalanceService服务实例，同时也只会启动一个mqClientFactory实例  
    3. 单点内所有的Consumer实例都会共用该实例对象。
    4. 每次定时触发mqClientFactory.doRebalance()都会对该JVM下持有的所有Consumer进行负载均衡
    5.  
    1. /**
    2.  * RebalanceService 线程默认每隔20s调用该方法
    3.  * ⚠️:每个 Java 服务单点只会启动一个 MQClientInstance 实例，单点内所有的 Consumer 实例都会持有该实例对象
    4.  * @see #registerConsumer Consumer 对象会将自己注册进 MQClientInstance
    5.  * @see #consumerTable Consumer对象注册表
    6.  *
    7.  * ⚠️:一个 Java 服务单点只有一个 RebalanceService 服务线程
    8.  * ⚠️:但每个 Consumer 实例都持有一个 RebalanceImpl 对象
    9.  */
    10. public void doRebalance() {
    11.     for (Map.Entry<String, MQConsumerInner> entry : this.consumerTable.entrySet()) {
    12.         MQConsumerInner impl = entry.getValue();
    13.         if (impl != null) {
    14.             try {
    15.                 impl.doRebalance();
    16.             } catch (Throwable e) {
    17.                 log.error("doRebalance exception", e);
    18.             }
    19.         }
    20.     }
    21. }
    22.

    主动触发

    1. 写这篇记录之时我还在思考，为什么刚刚那个定时任务不是交给JDK中的ScheduledExecutorService
    2. 事实上，RocketMQ中的很多定时任务也都是这么做的。
    3.  
    4. 可是直到刚刚我才明白，因为RebalanceService支持主动唤醒，提前执行任务。
    5.  
    6. Consumer上线时候触发主动负载均衡就是因为唤醒了RebalanceService线程，
    7. start()最后会调用rebalanceImmediately()
    8. public void rebalanceImmediately() {
    9.     this.rebalanceService.wakeup();
    10. }
    11.  
    12.

    Broker通知

    Broker下发通知指挥Consumer需要负载均衡则明显复杂很多，但万变不离其宗，无非是多几次Rpc调用而已，无非是网络传输了一遭而已。

    1. 每当DefaultMQPushConsumerImpl实例，调用start之后，总是会向Broker发送一个心跳
    2. 调用栈如下:
    3.     DefaultMQPushConsumerImpl.start()
    4. ->  MQClientInstance.sendHeartbeatToAllBrokerWithLock()
    5. ->  MQClientInstance.sendHeartbeatToAllBroker()
    6. ->  MQClientAPIImpl.sendHearbeat()
    7.

    Consumer启动之后会立即发出一个心跳包告知Broker。

    1. public int sendHearbeat(
    2.         String addr, HeartbeatData heartbeatData, long timeoutMillis
    3.     ) throws RemotingException, MQBrokerException, InterruptedException {
    4.         /* 又是一次 Rpc 远程调用 */
    5.         RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.HEART_BEAT, null);
    6.         request.setLanguage(clientConfig.getLanguage());
    7.         request.setBody(heartbeatData.encode());
    8.         RemotingCommand response = this.remotingClient.invokeSync(
    9.             addr, request, timeoutMillis
    10.         );
    11.     }
    12.

    根据RequestCode.HEART_BEAT得知此次Rpc的处理器为ClientManageProcessor ClientManageProcessor.heartBeat() -> ConsumerManager.registerConsumer()

    仅摘录关键代码:

    1. public boolean registerConsumer(String group, ClientChannelInfo clientChannelInfo,
    2.     ConsumeType consumeType, MessageModel messageModel, ConsumeFromWhere consumeFromWhere,
    3.     Set<SubscriptionData> subList, boolean isNotifyConsumerIdsChangedEnable
    4. ) {
    5.    /* 有新的Consumer上线则有新的SocketChannel建立 */
    6.    boolean r1 = consumerGroupInfo.updateChannel(
    7.        clientChannelInfo, consumeType, messageModel, consumeFromWhere
    8.    );
    9.  
    10.    /* 判断订阅信息是否发生变化 */
    11.    boolean r2 = consumerGroupInfo.updateSubscription(subList);
    12.  
    13.    if (r1 || r2) {
    14.        if (isNotifyConsumerIdsChangedEnable) {
    15.            /* 触发ConsumerGroupEvent.CHANGE事件 */
    16.            this.consumerIdsChangeListener.handle(ConsumerGroupEvent.CHANGE, group, consumerGroupInfo.getAllChannel());
    17.        }
    18.    } 
    19. }
    20.

    调用进行到DefaultConsumerIdsChangeListener.handle()
    可以看到如果是CHANGE事件则调用Broker2Client.notifyConsumerIdsChanged()

    1. public void notifyConsumerIdsChanged(Channel channel, String consumerGroup) {
    2.     /* 构造Rpc请求头 */
    3.     NotifyConsumerIdsChangedRequestHeader requestHeader = new NotifyConsumerIdsChangedRequestHeader();
    4.     requestHeader.setConsumerGroup(consumerGroup);
    5.     /* 构造Rpc请求对象 */
    6.     RemotingCommand request = RemotingCommand.createRequestCommand(
    7.         RequestCode.NOTIFY_CONSUMER_IDS_CHANGED, requestHeader
    8.     );
    9.  
    10.     /* 又是一次RPC */
    11.     this.brokerController.getRemotingServer().invokeOneway(channel, request, 10);
    12. }
    13.

    这个Rpc请求最终会交给ClientRemotingProcessor.notifyConsumerIdsChanged()处理

    1. public RemotingCommand notifyConsumerIdsChanged(ChannelHandlerContext ctx,
    2.     RemotingCommand request) {
    3.     NotifyConsumerIdsChangedRequestHeader requestHeader =(NotifyConsumerIdsChangedRequestHeader) request
    4.         .decodeCommandCustomHeader(NotifyConsumerIdsChangedRequestHeader.class);
    5.     /* 调用负载均衡逻辑 */
    6.     this.mqClientFactory.rebalanceImmediately();
    7. }
  • 相关阅读:
    SVG格式进行xss与xxe
    网络复习题带答案
    Cocos 3D开源游戏案例
    【Vue】 五分钟就能看懂的vue数据双向绑定
    安卓终端神器Termux （后面还有Termux常用的快捷键，妥妥干货分享，记得点赞收藏哦！）
    【办公自动化】在Excel中按条件筛选数据并存入新的表（文末送书）
    AC自动机笔记与例题整理
    阿里面试官问我Redis 数据结构，这一篇文章足以征服他
    单向环形链表构建(思路分析) [Java][数据结构]
    以色列战术耳机公司【Silynxcom】申请1000万美元纳斯达克IPO上市
  • 原文地址：https://blog.csdn.net/Java_ttcd/article/details/126462517