RocketMQ源码17-consumer 负载均衡（重平衡）

超、凢脫俗 2023-09-25 23:25 197阅读 0赞

前面我们在分析Consumer消费过程时，有提到一个非常重要的概念，就是重平衡，只有在经过重平衡后，消息的拉取对象PullMessageService才可以去Broker拉取消息，那么这篇文章就单独分析下什么是重平衡?

重平衡分析的前提是：集群消费模式。 重平衡要做的也很简单，就是给当前消费者分配属于它的逻辑消费队列。

1.什么是重平衡?

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

Rebalance机制的本意是为了提升消息的并行消费能力。例如，⼀个Topic下5个队列，在只有1个消费者的情况下，这个消费者将负责消费这5个队列的消息。如果此时我们增加⼀个消费者，那么就可以给其中⼀个消费者分配2个队列，给另⼀个分配3个队列，从而提升消息的并行消费能力。

由于⼀个队列最多分配给消费者组下的⼀个消费者，因此当某个消费者组下的消费者实例数量大于队列的数量时，多余的消费者实例将分配不到任何队列。

format_png

接下来以集群模式下的消息推模式DefaultMQPushConsumerImpl为例，看一下负载均衡的过程。

2. 消费者启动DefaultMQPushConsumerImpl

首先，消费者在启动时会做如下操作：

  1. 从NameServer更新当前消费者订阅主题的路由信息；
  2. 向Broker发送心跳，注册消费者；

  3. 唤醒负载均衡服务，触发一次负载均衡；

    public class DefaultMQPushConsumerImpl implements MQConsumerInner {

    1. public synchronized void start() throws MQClientException {
    2.     // ...
    3.     // 更新当前消费者订阅主题的路由信息
    4.     this.updateTopicSubscribeInfoWhenSubscriptionChanged();
    5.     this.mQClientFactory.checkClientInBroker();
    6.     // 向Broker发送心跳
    7.     this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
    8.     // 唤醒负载均衡服务
    9.     this.mQClientFactory.rebalanceImmediately();
    10. }

    }
    复制代码

2.1 更新主题路由信息

为了保证消费者拿到的主题路由信息是最新的（topic下有几个消息队列、消息队列的分布信息等），在进行负载均衡之前首先要更新主题的路由信息，在updateTopicSubscribeInfoWhenSubscriptionChanged方法中可以看到，首先获取了当前消费者订阅的所有主题信息（一个消费者可以订阅多个主题），然后进行遍历，向NameServer发送请求，更新每一个主题的路由信息，保证路由信息是最新的：

  1. public class DefaultMQPushConsumerImpl implements MQConsumerInner {
  2.     private void updateTopicSubscribeInfoWhenSubscriptionChanged() {
  3.         // 获取当前消费者订阅的主题信息
  4.         Map<String, SubscriptionData> subTable = this.getSubscriptionInner();
  5.         if (subTable != null) {
  6.             // 遍历订阅的主题信息
  7.             for (final Map.Entry<String, SubscriptionData> entry : subTable.entrySet()) {
  8.                 final String topic = entry.getKey();
  9.                 // 从NameServer更新主题的路由信息
  10.                 this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic);
  11.             }
  12.         }
  13.     }
  14. }
  15. 复制代码

2.2 注册消费者

2.2.1 发送心跳

由于Broker需要感知消费者数量的增减，所以每个消费者在启动的时候，会调用sendHeartbeatToAllBrokerWithLock向Broker发送心跳包，进行消费者注册：

  1. public class MQClientInstance {
  2.     public void sendHeartbeatToAllBrokerWithLock() {
  3.         if (this.lockHeartbeat.tryLock()) {
  4.             try {
  5.                 // todo 调用sendHeartbeatToAllBroker向Broker发送心跳
  6.                 this.sendHeartbeatToAllBroker();
  7.                 this.uploadFilterClassSource();
  8.             } catch (final Exception e) {
  9.                 log.error("sendHeartbeatToAllBroker exception", e);
  10.             } finally {
  11.                 this.lockHeartbeat.unlock();
  12.             }
  13.         } else {
  14.             log.warn("lock heartBeat, but failed. [{}]", this.clientId);
  15.         }
  16.     }
  17. }
  18. 复制代码

在sendHeartbeatToAllBroker方法中，可以看到从brokerAddrTable中获取了所有的Broker进行遍历（主从模式下也会向从节点发送请求注册），调用MQClientAPIImpl的sendHearbeat方法向每一个Broker发送心跳请求进行注册：

  1. public class MQClientInstance {
  2.     // Broker路由表
  3.     private final ConcurrentMap<String/* Broker Name */, HashMap<Long/* brokerId */, String/* address */>> brokerAddrTable =
  4.         new ConcurrentHashMap<String, HashMap<Long, String>>();
  5.     // 发送心跳
  6.     private void sendHeartbeatToAllBroker() {
  7.         final HeartbeatData heartbeatData = this.prepareHeartbeatData();
  8.         // ...
  9.         if (!this.brokerAddrTable.isEmpty()) {
  10.             long times = this.sendHeartbeatTimesTotal.getAndIncrement();
  11.             // 获取所有的Broker进行遍历， key为 Broker Name， value为同一个name下的所有Broker实例（主从模式下Broker的name一致）
  12.             Iterator<Entry<String, HashMap<Long, String>>> it = this.brokerAddrTable.entrySet().iterator();
  13.             while (it.hasNext()) {
  14.                 Entry<String, HashMap<Long, String>> entry = it.next();
  15.                 String brokerName = entry.getKey(); // broker name
  16.                 // 获取同一个Broker Name下的所有Broker实例
  17.                 HashMap<Long, String> oneTable = entry.getValue();
  18.                 if (oneTable != null) {
  19.                     // 遍历所有的实例
  20.                     for (Map.Entry<Long, String> entry1 : oneTable.entrySet()) {
  21.                         Long id = entry1.getKey();
  22.                         String addr = entry1.getValue();
  23.                         if (addr != null) { // 如果地址不为空
  24.                             // ...
  25.                             try {
  26.                                 // 发送心跳
  27.                                 int version = this.mQClientAPIImpl.sendHearbeat(addr, heartbeatData, clientConfig.getMqClientApiTimeout());
  28.                                 // ...
  29.                             } catch (Exception e) {
  30.                                 // ...
  31.                             }
  32.                         }
  33.                     }
  34.                 }
  35.             }
  36.         }
  37.     }
  38. }
  39. 复制代码

在MQClientAPIImpl的sendHearbeat方法中，可以看到构建了HEART_BEAT请求，然后向Broker发送：

  1. public class MQClientAPIImpl {
  2.    public int sendHearbeat(final String addr, final HeartbeatData heartbeatData, final long timeoutMillis
  3.     ) throws RemotingException, MQBrokerException, InterruptedException {
  4.         // 创建HEART_BEAT请求
  5.         RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.HEART_BEAT, null);
  6.         request.setLanguage(clientConfig.getLanguage());
  7.         request.setBody(heartbeatData.encode());
  8.         // 发送请求
  9.         RemotingCommand response = this.remotingClient.invokeSync(addr, request, timeoutMillis);
  10.         // ...
  11.     }
  12. }
  13. 复制代码

2.2.2 broker心跳请求处理

Broker在启动时注册了HEART_BEAT请求的处理器，可以看到请求处理器是ClientManageProcessor：

  1. public class BrokerController {
  2.     public void registerProcessor() {
  3.         ClientManageProcessor clientProcessor = new ClientManageProcessor(this);
  4.         // 注册HEART_BEAT请求的处理器ClientManageProcessor
  5.         this.remotingServer.registerProcessor(RequestCode.HEART_BEAT, clientProcessor, this.heartbeatExecutor);
  6.     }
  7. }
  8. 复制代码

进入到ClientManageProcessor的processRequest方法，如果请求是HEART_BEAT类型会调用heartBeat方法进行处理，这里也能看还有UNREGISTER_CLIENT类型的请求，从名字上可以看出是与取消注册有关的（这个稍后再说）：

  1. public class ClientManageProcessor extends AsyncNettyRequestProcessor implements NettyRequestProcessor {
  2.     @Override
  3.     public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request)
  4.         throws RemotingCommandException {
  5.         switch (request.getCode()) {
  6.             case RequestCode.HEART_BEAT: // 处理心跳请求
  7.                 return this.heartBeat(ctx, request);
  8.             case RequestCode.UNREGISTER_CLIENT: // 取消注册请求
  9.                 return this.unregisterClient(ctx, request);
  10.             case RequestCode.CHECK_CLIENT_CONFIG:
  11.                 return this.checkClientConfig(ctx, request);
  12.             default:
  13.                 break;
  14.         }
  15.         return null;
  16.     }
  17. }
  18. 复制代码

进入到heartBeat方法，可以看到，调用了ConsumerManager的registerConsumer注册消费者：

  1. public class ClientManageProcessor extends AsyncNettyRequestProcessor implements NettyRequestProcessor {
  2.     public RemotingCommand heartBeat(ChannelHandlerContext ctx, RemotingCommand request) {
  3.         // ...
  4.         for (ConsumerData data : heartbeatData.getConsumerDataSet()) {
  5.             // ...
  6.             // 注册Consumer
  7.             boolean changed = this.brokerController.getConsumerManager().registerConsumer(
  8.                 data.getGroupName(), clientChannelInfo, data.getConsumeType(), data.getMessageModel(), data.getConsumeFromWhere(),
  9.                 data.getSubscriptionDataSet(), isNotifyConsumerIdsChangedEnable);
  10.             // ...
  11.         }
  12.         // ...
  13.         return response;
  14.     }
  15. }
  16. 复制代码

进行注册

ConsumerManager的registerConsumer方法的理逻辑如下：

  1. 根据组名称获取该消费者组的信息ConsumerGroupInfo对象。如果获取为空，会创建一个ConsumerGroupInfo，记录了消费者组的相关信息；
  2. 判断消费者是否发生了变更，如果如果发生了变化，会触发CHANGE变更事件（这个稍后再看）；

  3. 触发REGISTER注册事件；

    public class ConsumerManager {

    1. public boolean registerConsumer(final String group, final ClientChannelInfo clientChannelInfo,
    2.     ConsumeType consumeType, MessageModel messageModel, ConsumeFromWhere consumeFromWhere,
    3.     final Set<SubscriptionData> subList, boolean isNotifyConsumerIdsChangedEnable) {
    4.     // 根据组名称获取消费者组信息
    5.     ConsumerGroupInfo consumerGroupInfo = this.consumerTable.get(group);
    6.     if (null == consumerGroupInfo) { // 如果为空新增ConsumerGroupInfo对象
    7.         ConsumerGroupInfo tmp = new ConsumerGroupInfo(group, consumeType, messageModel, consumeFromWhere);
    8.         ConsumerGroupInfo prev = this.consumerTable.putIfAbsent(group, tmp);
    9.         consumerGroupInfo = prev != null ? prev : tmp;
    10.     }
    11.     boolean r1 =
    12.         consumerGroupInfo.updateChannel(clientChannelInfo, consumeType, messageModel,
    13.             consumeFromWhere);
    14.     boolean r2 = consumerGroupInfo.updateSubscription(subList);
    15.     // 如果有变更
    16.     if (r1 || r2) {
    17.         if (isNotifyConsumerIdsChangedEnable) {
    18.             // 通知变更
    19.             this.consumerIdsChangeListener.handle(ConsumerGroupEvent.CHANGE, group, consumerGroupInfo.getAllChannel());
    20.         }
    21.     }
    22.     // 注册Consumer
    23.     this.consumerIdsChangeListener.handle(ConsumerGroupEvent.REGISTER, group, subList);
    24.     return r1 || r2;
    25. }

    }
    复制代码

进入到DefaultConsumerIdsChangeListener的handle方法中，可以看到如果是REGISTER事件，会通过ConsumerFilterManager的register方法进行注册，注册的详细过程这里先不展开讲解：

  1. public class DefaultConsumerIdsChangeListener implements ConsumerIdsChangeListener {
  2.     @Override
  3.     public void handle(ConsumerGroupEvent event, String group, Object... args) {
  4.         if (event == null) {
  5.             return;
  6.         }
  7.         switch (event) {
  8.             case CHANGE:// 如果是消费者变更事件
  9.                 // ...
  10.                 break;
  11.             case UNREGISTER: // 如果是取消注册事件
  12.                 this.brokerController.getConsumerFilterManager().unRegister(group);
  13.                 break;
  14.             case REGISTER: // 如果是注册事件
  15.                 if (args == null || args.length < 1) {
  16.                     return;
  17.                 }
  18.                 Collection<SubscriptionData> subscriptionDataList = (Collection<SubscriptionData>) args[0];
  19.                 // 进行注册
  20.                 this.brokerController.getConsumerFilterManager().register(group, subscriptionDataList);
  21.                 break;
  22.             default:
  23.                 throw new RuntimeException("Unknown event " + event);
  24.         }
  25.     }
  26. }
  27. 复制代码

2.3 立即重平衡

org.apache.rocketmq.client.impl.factory.MQClientInstance#rebalanceImmediately：

  1. public void rebalanceImmediately() {
  2.     this.rebalanceService.wakeup();
  3. }
  4. 复制代码

这里会唤醒org.apache.rocketmq.client.impl.consumer.RebalanceService的run方法中的 this.mqClientFactory.doRebalance();:

  1. @Override
  2. public void run() {
  3.     log.info(this.getServiceName() + " service started");
  6.     while (!this.isStopped()) {
  7.         // todo 等待20s执行一次 内部使用了juc的CountDownLatch, 使得这里启动后仍然是阻塞的
  8.         this.waitForRunning(waitInterval);
  9.         // todo
  10.         this.mqClientFactory.doRebalance();
  11.     }
  13.     log.info(this.getServiceName() + " service end");
  14. }
  15. 复制代码

接下来我们重点看一下 重平衡。

3.重平衡之queue的分配

rocketmq中的rebalance是consumer实例自身完成的

  1. public class RebalanceService extends ServiceThread {
  2.     private static long waitInterval = Long.parseLong(System.getProperty(
  3.             "rocketmq.client.rebalance.waitInterval", "20000"));
  5.     //TODO:...略.....        
  7.     @Override
  8.     public void run() {
  9.         log.info(this.getServiceName() + " service started");
  11.         while (!this.isStopped()) {
  12.             this.waitForRunning(waitInterval);
  13.             this.mqClientFactory.doRebalance();
  14.         }
  16.         log.info(this.getServiceName() + " service end");
  17.     }
  18. }
  19. 复制代码

从代码中可以看到，它默认是每隔20s触发一次重平衡。

那么我们继续看下consumer客户端是如何完成重平衡的：

org.apache.rocketmq.client.impl.consumer.RebalanceImpl#doRebalance:

  1. public void doRebalance(final boolean isOrder) {
  2.     Map<String, SubscriptionData> subTable = this.getSubscriptionInner();
  3.     if (subTable != null) {
  4.         for (final Map.Entry<String, SubscriptionData> entry : subTable.entrySet()) {
  5.             final String topic = entry.getKey();
  6.             try {
  7.                 // todo 客户端负载均衡：根据主题来处理负载均衡
  8.                 this.rebalanceByTopic(topic, isOrder);
  9.             } catch (Throwable e) {
  10.                 if (!topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
  11.                     log.warn("rebalanceByTopic Exception", e);
  12.                 }
  13.             }
  14.         }
  15.     }
  17.     this.truncateMessageQueueNotMyTopic();
  18. }
  19. 复制代码

我们继续往下走： org.apache.rocketmq.client.impl.consumer.RebalanceImpl#rebalanceByTopic:

  1. private void rebalanceByTopic(final String topic, final boolean isOrder) {
  2.     switch (messageModel) {
  3.         // 广播模式：不需要处理负载均衡，每个消费者都要消费，只需要更新负载信息
  4.         case BROADCASTING: {
  5.             // 更新负载均衡信息，这里传入的参数是mqSet，即所有队列
  6.             Set<MessageQueue> mqSet = this.topicSubscribeInfoTable.get(topic);
  7.             if (mqSet != null) {
  8.                 boolean changed = this.updateProcessQueueTableInRebalance(topic, mqSet, isOrder);
  9.                 if (changed) {
  10.                     this.messageQueueChanged(topic, mqSet, mqSet);
  11.                     log.info("messageQueueChanged {} {} {} {}",
  12.                         consumerGroup,
  13.                         topic,
  14.                         mqSet,
  15.                         mqSet);
  16.                 }
  17.             } else {
  18.                 log.warn("doRebalance, {}, but the topic[{}] not exist.", consumerGroup, topic);
  19.             }
  20.             break;
  21.         }
  22.         // todo 集群模式
  23.         case CLUSTERING: {
  24.             // 从主题订阅信息缓存表中获取主题的队列信息, 获取这个topic下的所有队列（默认是4个）
  25.             Set<MessageQueue> mqSet = this.topicSubscribeInfoTable.get(topic);
  26.             //发送请求从Broker中获取该消费组内当前所有的消费者客户端ID，主题的队
  27.             //列可能分布在多个Broker上，那么请求该发往哪个Broker呢？
  28.             //RocketeMQ从主题的路由信息表中随机选择一个Broker
  29.             List<String> cidAll = this.mQClientFactory.findConsumerIdList(topic, consumerGroup);
  30.             if (null == mqSet) {
  31.                 if (!topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
  32.                     log.warn("doRebalance, {}, but the topic[{}] not exist.", consumerGroup, topic);
  33.                 }
  34.             }
  36.             if (null == cidAll) {
  37.                 log.warn("doRebalance, {} {}, get consumer id list failed", consumerGroup, topic);
  38.             }
  40.             // 如果mqSet、cidAll任意一个为空，则忽略本次消息队列负载
  41.             if (mqSet != null && cidAll != null) {
  42.                 List<MessageQueue> mqAll = new ArrayList<MessageQueue>();
  43.                 mqAll.addAll(mqSet);
  45.                 // 对cidAll、mqAll进行排序
  46.                 // 这一步很重要，同一个消费组内看到的视图应保持一致，确保同一个消费队列不会被多个消费者分配
  47.                 Collections.sort(mqAll);
  48.                 Collections.sort(cidAll);
  50.                 // 默认是平均分配策略
  51.                 AllocateMessageQueueStrategy strategy = this.allocateMessageQueueStrategy;
  53.                 List<MessageQueue> allocateResult = null;
  54.                 try {
  55.                     // todo 分配算法
  56.                     allocateResult = strategy.allocate(
  57.                         this.consumerGroup,
  58.                         this.mQClientFactory.getClientId(),
  59.                         mqAll,
  60.                         cidAll);
  61.                 } catch (Throwable e) {
  62.                     log.error("AllocateMessageQueueStrategy.allocate Exception. allocateMessageQueueStrategyName={}", strategy.getName(),
  63.                         e);
  64.                     return;
  65.                 }
  67.                 //TODO:保存了当前消费者需要消费的队列
  68.                 Set<MessageQueue> allocateResultSet = new HashSet<MessageQueue>();
  69.                 if (allocateResult != null) {
  70.                     allocateResultSet.addAll(allocateResult);
  71.                 }
  74.                 // todo 对比消息队列是否发生变化 更新负载均衡信息，传入参数是 allocateResultSet，即当前consumer分配到的队列
  75.                 boolean changed = this.updateProcessQueueTableInRebalance(topic, allocateResultSet, isOrder);
  76.                 if (changed) {
  77.                     log.info(
  78.                         "rebalanced result changed. allocateMessageQueueStrategyName={}, group={}, topic={}, clientId={}, mqAllSize={}, cidAllSize={}, rebalanceResultSize={}, rebalanceResultSet={}",
  79.                         strategy.getName(), consumerGroup, topic, this.mQClientFactory.getClientId(), mqSet.size(), cidAll.size(),
  80.                         allocateResultSet.size(), allocateResultSet);
  81.                     this.messageQueueChanged(topic, mqSet, allocateResultSet);
  82.                 }
  83.             }
  84.             break;
  85.         }
  86.         default:
  87.             break;
  88.     }
  89. }
  90. 复制代码

这里根据消费模式进行了判断

  1. 广播模式：消费者要消费所有队列的数据，所以如果queue变动了，就更新消费者的消费队列，总之就是消费者要消费所有队列。
  2. 集群模式：相同Consumer Group的每个Consumer实例平均分摊同一个Topic的消息。即每条消息只会被发送到Consumer Group中的某个Consumer。

我们这里就关注集群模式。

在继续探讨重平衡之前，先提一下queue的分配算法：

  1. 平均分配策略(默认）

它会根据当前消费者在消费者集群中的位置(索引),以及queueCount%consumerCount的余数，计算当前消费者可以分配的queue。

举例：如果是4个queue,3个消费者; consumer1分配queueid=0,1;consumer2分配queueid=2,consumer3分配queueid=3。

  1. 环形平均策略

环形平均策略根据消费者的顺序，依次在由queue队列组成的环形图中逐个分配；该算法不用事先计算每个Consumer需要分配几个Queue，直接一个一个分即可。

举例：如果是4个queue,3个消费者; consumer1分配queueid=0,3;consumer2分配queueid=1,consumer3分配queueid=2。它和平均分配不同的是，它先给consumer1分配queueid=0,consumer2分配queueid=1,consumer3分配queueid=2,然后再将多余的queueid=3分配给consumer1。

  1. 一致性hash策略

该策略会将consumer的hash值作为Node节点存放到hash环上，然后将queue的hash值也放到hash环上，通过顺时针方向，距离queue最近的那个consumer就是该queue要分配的consumer。其可以有效减少由于消费者组扩容或缩容所带来的大量的Rebalance

  1. 同机房策略

该算法会根据queue的部署机房位置和consumer的位置，过滤出当前consumer相同机房的queue。不过，这种情况下，brokerName在命名上要按照的机房名@brokerName格式。

  1. 就近机房策略

它是上面同机房策略的补充。比如同机房中只有queue而没有consumer,此时就可以使用就近机房策略进行补充。

对应的接口：

  1. /**
  2.  * queue分配策略
  3.  */
  4. public interface AllocateMessageQueueStrategy {
  6.     /**
  7.      * Allocating by consumer id
  8.      *
  9.      * @param consumerGroup:消费者组
  10.      * @param currentCID:当前的消费者id
  11.      * @param mqAll:topic下的所有队列
  12.      * @param cidAl：消费者组下的所有消费者id
  13.      * @return :返回当前消费者需要消费的队列
  14.      */
  15.     List<MessageQueue> allocate(
  16.         final String consumerGroup,
  17.         final String currentCID,
  18.         final List<MessageQueue> mqAll,
  19.         final List<String> cidAll
  20.     );
  21. }
  22. 复制代码

默认使用的是平均分配策略 AllocateMessageQueueAveragely

这里我们重点来分析平均负载策略AllocateMessageQueueAveragely：

  1. public List<MessageQueue> allocate(String consumerGroup, String currentCID, 
  2.         List<MessageQueue> mqAll, List<String> cidAll) {
  3.     // 返回值        
  4.     List<MessageQueue> result = new ArrayList<MessageQueue>();
  6.     // 省略一些判断操作
  7.     ...
  10.     int index = cidAll.indexOf(currentCID);
  11.     int mod = mqAll.size() % cidAll.size();
  12.     // 1. 消费者数量大于队列数量：averageSize = 1
  13.     // 2. 消费者数量小于等于队列数量：averageSize = 队列数量 / 消费者数量，还要处理个+1的操作
  14.     int averageSize = mqAll.size() <= cidAll.size() 
  15.         ? 1 : (mod > 0 && index < mod 
  16.             ? mqAll.size() / cidAll.size() + 1 : mqAll.size() / cidAll.size());
  17.     int startIndex = (mod > 0 && index < mod) ? index * averageSize : index * averageSize + mod;
  18.     int range = Math.min(averageSize, mqAll.size() - startIndex);
  19.     for (int i = 0; i < range; i++) {
  20.         result.add(mqAll.get((startIndex + i) % mqAll.size()));
  21.     }
  22.     return result;
  23. }
  24. 复制代码

这个方法中，关键的分配方法就在后面几行，如果只看代码，会感觉有点晕，这里我举一个例子来简单解释下：

假设：messageQueue一共有6个，consumer有4个，当前consumer的index为1，有了这些前提后，接下来我们就来看它的分配过程了。

  1. 计算取余操作：6 % 4 = 2，这表明messageQueue不能平均分配给每个consumer，接下来就来看看这个余数2是如何处理的

  2. 计算每个consumer平均处理的messageQueue数量

    • 这里需要注意，如果consumer数量大于messageQueue数量，那每个consumer最多只会分配到一个messageQueue，这种情况下，余数2不会进行处理，并且有的consumer处理的messageQueue数量为0，同一个messageQueue不会同时被两个及以上的consumer消费掉
    • 这里的messageQueue数量为6，consumer为4，计算得到每个consumer处理的队列数最少为1，除此之外，为了实现“平均”，有2个consumer会需要多处理1个messageQueue，按“平均”的分配原则，如果index小于mod，则会分配多1个messageQueue，这里的mod为2，结果如下：



















    消费者索引 0 1 2 3
    处理数量 2 2 1 1

  3. 分配完每个consumer处理的messageQueue数量后，这些messageQueue该如何分配呢？从代码来看，分配时会先分配完一个consumer，再分配下一个consumer，最终结果就是这样：
























    队列 Q0 Q1 Q2 Q3 Q4 Q5
    消费者 C1 C1 C2 C2 C4 C5

从图中可以看到，在6个messageQueue、4个consumer、当前consumer的index为1的情况下，当前consumer会分到2个队列，分别为Q2/Q3.

其他分配策略 大家可自行阅读源码

4.重平衡核心逻辑分析

  1. /**
  2.  * 对比消息队列是否发生变化，主要思路是遍历当前负载队列集
  3.  * 合，如果队列不在新分配队列的集合中，需要将该队列停止消费并保
  4.  * 存消费进度；遍历已分配的队列，如果队列不在队列负载表中
  5.  * （processQueueTable），则需要创建该队列拉取任务PullRequest，
  6.  * 然后添加到PullMessageService线程的pullRequestQueue中，
  7.  * PullMessageService才会继续拉取任务
  8.  */
  9. private boolean updateProcessQueueTableInRebalance(final String topic, final Set<MessageQueue> mqSet,
  10.     final boolean isOrder) {
  11.     boolean changed = false;
  13.     /**
  14.      * ConcurrentMap〈MessageQueue, ProcessQueue〉
  15.      * processQueueTable是当前消费者负载的消息队列缓存表，如果缓存表
  16.      * 中的MessageQueue不包含在mqSet中，说明经过本次消息队列负载后，
  17.      * 该mq被分配给其他消费者，需要暂停该消息队列消息的消费。方法是
  18.      * 将ProccessQueue的状态设置为droped=true，该ProcessQueue中的消
  19.      * 息将不会再被消费，调用removeUnnecessaryMessageQueue方法判断是
  20.      * 否将MessageQueue、ProccessQueue从缓存表中移除。
  21.      * removeUnnecessaryMessageQueue在RebalanceImple中定义为抽象方
  22.      * 法。removeUnnecessaryMessageQueue方法主要用于持久化待移除
  23.      * MessageQueue的消息消费进度。在推模式下，如果是集群模式并且是
  24.      * 顺序消息消费，还需要先解锁队列
  25.      */
  26.     Iterator<Entry<MessageQueue, ProcessQueue>> it = this.processQueueTable.entrySet().iterator();
  27.     while (it.hasNext()) {
  28.         Entry<MessageQueue, ProcessQueue> next = it.next();
  29.         MessageQueue mq = next.getKey();
  30.         ProcessQueue pq = next.getValue();
  32.         if (mq.getTopic().equals(topic)) {
  33.             if (!mqSet.contains(mq)) {
  34.                 pq.setDropped(true);
  35.                 if (this.removeUnnecessaryMessageQueue(mq, pq)) {
  36.                     it.remove();
  37.                     changed = true;
  38.                     log.info("doRebalance, {}, remove unnecessary mq, {}", consumerGroup, mq);
  39.                 }
  40.             } else if (pq.isPullExpired()) {
  41.                 switch (this.consumeType()) {
  42.                     case CONSUME_ACTIVELY:
  43.                         break;
  44.                     case CONSUME_PASSIVELY:
  45.                         pq.setDropped(true);
  46.                         if (this.removeUnnecessaryMessageQueue(mq, pq)) {
  47.                             it.remove();
  48.                             changed = true;
  49.                             log.error("[BUG]doRebalance, {}, remove unnecessary mq, {}, because pull is pause, so try to fixed it",
  50.                                 consumerGroup, mq);
  51.                         }
  52.                         break;
  53.                     default:
  54.                         break;
  55.                 }
  56.             }
  57.         }
  58.     }
  60.     /**
  61.      * 遍历本次负载分配到的队列集合，如果
  62.      * processQueueTable中没有包含该消息队列，表明这是本次新增加的消
  63.      * 息队列，首先从内存中移除该消息队列的消费进度，然后从磁盘中读
  64.      * 取该消息队列的消费进度，创建PullRequest对象。这里有一个关键，
  65.      * 如果读取到的消费进度小于0，则需要校对消费进度。RocketMQ提供了
  66.      * CONSUME_FROM_LAST_OFFSET、CONSUME_FROM_FIRST_OFFSET、
  67.      * CONSUME_FROM_TIMESTAMP方式，在创建消费者时可以通过调用
  68.      * DefaultMQPushConsumer#setConsumeFromWhere方法进行设置
  69.      */
  70.     List<PullRequest> pullRequestList = new ArrayList<PullRequest>();
  71.     for (MessageQueue mq : mqSet) {
  72.         if (!this.processQueueTable.containsKey(mq)) {
  74.             /**
  75.              * 经过消息队列重新负载（分配）后，分配到新的消息队列时，首
  76.              * 先需要尝试向Broker发起锁定该消息队列的请求，如果返回加锁成
  77.              * 功，则创建该消息队列的拉取任务，否则跳过，等待其他消费者释放
  78.              * 该消息队列的锁，然后在下一次队列重新负载时再尝试加锁
  79.              */
  80.             // 顺序消息
  81.             if (isOrder && !this.lock(mq)) {
  82.                 log.warn("doRebalance, {}, add a new mq failed, {}, because lock failed", consumerGroup, mq);
  83.                 continue;
  84.             }
  86.             this.removeDirtyOffset(mq);
  87.             ProcessQueue pq = new ProcessQueue();
  88.             // todo PullRequest的nextOffset计算逻辑位于RebalancePushImpl#computePullFromWhere
  89.             long nextOffset = this.computePullFromWhere(mq);
  90.             if (nextOffset >= 0) {
  91.                 ProcessQueue pre = this.processQueueTable.putIfAbsent(mq, pq);
  92.                 if (pre != null) {
  93.                     log.info("doRebalance, {}, mq already exists, {}", consumerGroup, mq);
  94.                 } else {
  95.                     log.info("doRebalance, {}, add a new mq, {}", consumerGroup, mq);
  96.                     PullRequest pullRequest = new PullRequest();
  97.                     pullRequest.setConsumerGroup(consumerGroup);
  98.                     pullRequest.setNextOffset(nextOffset);
  99.                     pullRequest.setMessageQueue(mq);
  100.                     pullRequest.setProcessQueue(pq);
  101.                     pullRequestList.add(pullRequest);
  102.                     changed = true;
  103.                 }
  104.             } else {
  105.                 log.warn("doRebalance, {}, add new mq failed, {}", consumerGroup, mq);
  106.             }
  107.         }
  108.     }
  110.     // todo 将PullRequest加入PullMessageService，以便唤醒PullMessageService线程
  111.     this.dispatchPullRequest(pullRequestList);
  113.     return changed;
  114. }
  115. 复制代码
  1. 先将分配到的消息队列集合（mqSet）与processQueueTable做一个过滤比对

format_png 1

  • 上图中processQueueTable标注的红色部分，表示与分配到的消息队列集合mqSet互不包含。将这些队列设置dropped属性为true，然后查看这些队列是否可以移除出processQueueTable缓存变量，这里具体执行removeUnnecessaryMessageQueue()方法。
  • 上图中processQueueTable的绿色部分，表示与分配到的消息队列集合mqSet的交集。判断该ProcessQueue是否已经过期了，在Pull模式的不用管，如果是Push模式的，设置dropped属性为true，并且调用removeUnnecessaryMessageQueue()方法，像上面一样尝试移除Entry。

dropped属性很重要，后面讲重复消费消费暂停都会提到它。

  1. 为过滤后的消息队列集合（mqSet）中的每个MessageQueue创建一个ProcessQueue对象并存入RebalanceImpl的processQueueTable队列中（其中调用RebalanceImpl实例的computePullFromWhere(MessageQueue mq)方法获取该MessageQueue对象的下一个进度消费值offset，随后填充至接下来要创建的pullRequest对象属性中），并创建拉取请求对象—pullRequest添加到拉取列表—pullRequestList中，最后执行dispatchPullRequest()方法，将Pull消息的请求对象PullRequest依次放入PullMessageService服务线程的阻塞队列pullRequestQueue中，待该服务线程取出后向Broker端发起Pull消息的请求。

消息消费队列在同一消费组不同消费者之间的负载均衡，其核心设计理念是在一个消息消费队列在同一时间只允许被同一消费组内的一个消费者消费，一个消息消费者能同时消费多个消息队列

关于上面的核心逻辑方法我在举个栗子：假设有两个消费者c1,c2； 队列默认是4个。
消费者c1第一次重平衡：

  1. 消费者c1先启动，由于只有一个消费者，所以分配给c1的队列是4个(mqSet.size()=4)；
  2. 然后开始处理重平衡逻辑，第一次进来 processQueueTable肯定是empty的，所以先跳过遍历逻辑。

  3. 遍历分配到的mqSet集合（遍历4次）

    • 创建ProcessQueue对象
    • 读取当前MessageQueue的偏移量,去broker中读取
    • 将当前MessageQueue和ProcessQueue 放入processQueueTable中
    • 创建PullRequest对象，设置队列MessageQueue，缓存消息队列ProcessQueue，以及当前MessageQueue的偏移量，然后放入pullRequestList集合中

  4. 然后分发pullRequestList，将拉取消息的请求对象PullRequest依次放入PullMessageService服务线程的阻塞队列pullRequestQueue中，待该服务线程取出后向Broker端发起拉取消息的请求。

消费者c2重平衡：

  1. 消费者c2启动，先不关注它重平衡的逻辑，就假设它分配了2个队列

消费者c1第二次重平衡

  1. 消费者c1再次重平衡，经过平均分配算法后，c1也分配了2个queue(因为c2分配了2个，他们平均分),所以mqSet.size()=2

  2. 然后开始处理重平衡逻辑,由于消费者c1第一次重平衡时，processQueueTable已经放入了4个queue(MessageQueue),所以这里开始遍历。如果mqSet(这里是2个)不包含当前entry的key(MessageQueue)，则：

    • 将当前entry中的ProcessQueue的属性 dropped 设置为true.
    • 从RemoteBrokerOffsetStore#offsetTable属性中，移除当前queue(MessageQueue).因为这个队列已经部署了c1了。
    • 将当前entry从processQueueTable中移除

  3. 这样遍历完，processQueueTable 还剩下两个entry（这两个entry的key是在mqSet集合中的)

  4. 继续遍历mqSet(2个queue),但是继续判断(!processQueueTable.containsKey(mq)),这里肯定是false,因为前面分析过了，它是还剩下2个MessageQueue,所以跳过if,最终装PullRequest的容器集合是empty的,所以就没有以后了。。。。 不难发现，c1在第2次重平衡后，并没有构建全新的pullRequestList，所以dispatchPullRequest(pullRequestList)方法内部不会做任何逻辑。这样c1还是有4个PullRequest, 那么第2次重平衡好像就没有什么用处了？

    • 不是的，这就是前面提到的一个非常重要的属性设置dropped=true

    • 在拉取消息时，首先就会判断这个属性

      public void pullMessage(final PullRequest pullRequest) {

      1. final ProcessQueue processQueue = pullRequest.getProcessQueue();
      2. //TODO:如果是dropped=true,则本次拉取请求直接丢弃，不会在将PullRequest放回阻塞队列中
      3. if (processQueue.isDropped()) {
      4.     log.info("the pull request[{}] is dropped.", pullRequest.toString());
      5.     return;
      6. }
      8. //TODO:.......

      }
      复制代码

如果是dropped=true,则本次拉取请求直接丢弃，不会在将PullRequest放回阻塞队列中,这样c1就只消费2个队列了。所以在消息拉取之前，首先判断该队列是否被丢弃，如果已丢弃，则直接放弃本次拉取任务。

5.重平衡导致的重复消费

我们直接看消费相关的代码：

ConsumeMessageConcurrentlyService

  1. public void run() {
  2.     /**
  3.      * 先检查processQueue的dropped，如果设置为true，则停止该队列的消费。在进行消息重新负
  4.      * 载时，如果该消息队列被分配给消费组内的其他消费者，需要将
  5.      * droped设置为true，阻止消费者继续消费不属于自己的消息队列
  6.      */
  7.     if (this.processQueue.isDropped()) {
  8.         log.info("the message queue not be able to consume, because it's dropped. group={} {}", ConsumeMessageConcurrentlyService.this.consumerGroup, this.messageQueue);
  9.         return;
  10.     }
  12.     // 取出消息监听器 获取消息监听器MessageListenerConcurrently
  13.     MessageListenerConcurrently listener = ConsumeMessageConcurrentlyService.this.messageListener;
  14.     ConsumeConcurrentlyContext context = new ConsumeConcurrentlyContext(messageQueue);
  15.     ConsumeConcurrentlyStatus status = null;
  16.     /**
  17.      * 恢复重试消息主题名。这是为什么呢？这是由消息重试
  18.      * 机制决定的，RocketMQ将消息存入CommitLog文件时，如果发现消息的
  19.      * 延时级别delayTimeLevel大于0，会先将重试主题存入消息的属性，然
  20.      * 后将主题名称设置为SCHEDULE_TOPIC_XXXX，以便之后重新参与消息消
  21.      * 费
  22.      */
  23.     defaultMQPushConsumerImpl.resetRetryAndNamespace(msgs, defaultMQPushConsumer.getConsumerGroup());
  25.     ...
  26.     try {
  27.         ...
  28.         // todo 执行具体的消息消费，调用应用程序消息监听器的
  29.         //consumeMessage方法，进入具体的消息消费业务逻辑，返回该批消息
  30.         //的消费结果，即CONSUME_SUCCESS（消费成功）或
  31.         //RECONSUME_LATER（需要重新消费）
  32.         status = listener.consumeMessage(Collections.unmodifiableList(msgs), context);
  33.     } 
  34.     ...
  36.     ...
  38.     /**
  39.      * 执行业务消息消费后，在处理结果前再次验证一次
  40.      * ProcessQueue的isDropped状态值。如果状态值为true，将不对结果进
  41.      * 行任何处理。也就是说，在消息消费进入第四步时，如果因新的消费
  42.      * 者加入或原先的消费者出现宕机，导致原先分配给消费者的队列在负
  43.      * 载之后分配给了别的消费者，那么消息会被重复消费
  44.      */
  45.     if (!processQueue.isDropped()) {
  46.         // todo
  47.         ConsumeMessageConcurrentlyService.this.processConsumeResult(status, context, this);
  48.     } else {
  49.         log.warn("processQueue is dropped without process consume result. messageQueue={}, msgs={}", messageQueue, msgs);
  50.     }
  51. }
  52. 复制代码

而导致重复消费的就是在处理消费结果这里：

  1. //TODO:可能会导致重复消费
  2. if (!processQueue.isDropped()) {
  3.     ConsumeMessageConcurrentlyService.this.processConsumeResult(status, context, this);
  4. } else {
  5.     log.warn("processQueue is dropped without process consume result. messageQueue={}, msgs={}", messageQueue, msgs);
  6. }
  7. 复制代码

因为走到这里，说明消费者已经消费完了，但是可能由于重平衡可能导致我现在正消费的队列dropped=true，所以它不会去处理消费结果，而不处理消费结果就是不会去更新消费偏移量，这样同一个消费者组里的其他消费者可能还会消费到这条消息。这样就产生了重复消费。

注意：一条消息，被同一个消费者组里的不同消费者都消费过，这也是重复消费。

6. Rebalance的触发

6.1 消费者启动时触发

在文章开头已经讲过，消费者在启动时会进行一次负载均衡，这里便不再赘述。

6.2 消费者变更时触发

在消费者注册时讲到，如果发现消费者有变更会触发变更事件，当处于以下两种情况之一时会被判断为消费者发生了变化，需要进行负载均衡：

  • 当前注册的消费者对应的Channel对象之前不存在；

  • 当前注册的消费者订阅的主题信息发生了变化，也就是消费者订阅的主题有新增或者删除；

    public class ConsumerManager {

    1. /**
    2.  *  注册消费者
    3.  * @param group 消费者组名称
    4.  * @param clientChannelInfo 注册的消费者对应的Channel信息
    5.  * @param consumeType 消费类型
    6.  * @param messageModel 
    7.  * @param consumeFromWhere 消费消息的位置
    8.  * @param subList 消费者订阅的主题信息
    9.  * @param isNotifyConsumerIdsChangedEnable 是否通知变更
    10.  * @return
    11.  */
    12. public boolean registerConsumer(final String group, final ClientChannelInfo clientChannelInfo,
    13.     ConsumeType consumeType, MessageModel messageModel, ConsumeFromWhere consumeFromWhere,
    14.     final Set<SubscriptionData> subList, boolean isNotifyConsumerIdsChangedEnable) {
    15.     // 根据组名称获取消费者组信息
    16.     ConsumerGroupInfo consumerGroupInfo = this.consumerTable.get(group);
    17.     if (null == consumerGroupInfo) { // 如果为空新增
    18.         ConsumerGroupInfo tmp = new ConsumerGroupInfo(group, consumeType, messageModel, consumeFromWhere);
    19.         ConsumerGroupInfo prev = this.consumerTable.putIfAbsent(group, tmp);
    20.         consumerGroupInfo = prev != null ? prev : tmp;
    21.     }
    22.     // 更新Channel
    23.     boolean r1 =
    24.         consumerGroupInfo.updateChannel(clientChannelInfo, consumeType, messageModel,
    25.             consumeFromWhere);
    26.     // 更新订阅信息
    27.     boolean r2 = consumerGroupInfo.updateSubscription(subList);
    28.     // 如果有变更
    29.     if (r1 || r2) {
    30.         if (isNotifyConsumerIdsChangedEnable) {
    31.             // 通知变更，consumerGroupInfo中存储了该消费者组下的所有消费者的channel
    32.             this.consumerIdsChangeListener.handle(ConsumerGroupEvent.CHANGE, group, consumerGroupInfo.getAllChannel());
    33.         }
    34.     }
    35.     // 注册Consumer
    36.     this.consumerIdsChangeListener.handle(ConsumerGroupEvent.REGISTER, group, subList);
    37.     return r1 || r2;
    38. }

    }
    复制代码

6.2.1 Channel变更

在updateChannel方法中，首先将变更状态updated初始化为false，然后根据消费者的channel从channelInfoTable路由表中获取对应的ClientChannelInfo对象：

  • 如果ClientChannelInfo对象获取为空，表示之前不存在该消费者的channel信息，将其加入到路由表中，变更状态置为true，表示消费者有变化；
  • 如果获取不为空，判断clientid是否一致，如果不一致更新为最新的channel信息，但是变更状态updated不发生变化；

也就是说，如果注册的消费者之前不存在，那么将变更状态置为true，表示消费者数量发生了变化。

  1. // key为消费者对应的channle，value为chanel信息
  2.    private final ConcurrentMap<Channel, ClientChannelInfo> channelInfoTable =
  3.         new ConcurrentHashMap<Channel, ClientChannelInfo>(16);
  5.    public boolean updateChannel(final ClientChannelInfo infoNew, ConsumeType consumeType,
  6.         MessageModel messageModel, ConsumeFromWhere consumeFromWhere) {
  7.         boolean updated = false; // 变更状态初始化为false
  8.         this.consumeType = consumeType;
  9.         this.messageModel = messageModel;
  10.         this.consumeFromWhere = consumeFromWhere;
  11.         // 从channelInfoTable中获取对应的Channel信息, 
  12.         ClientChannelInfo infoOld = this.channelInfoTable.get(infoNew.getChannel());
  13.         if (null == infoOld) { // 如果为空
  14.             // 新增
  15.             ClientChannelInfo prev = this.channelInfoTable.put(infoNew.getChannel(), infoNew);
  16.             if (null == prev) { // 如果之前不存在
  17.                 log.info("new consumer connected, group: {} {} {} channel: {}", this.groupName, consumeType,
  18.                     messageModel, infoNew.toString());
  19.                 // 变更状态置为true
  20.                 updated = true;
  21.             }
  22.             infoOld = infoNew;
  23.         } else {
  24.             // 如果之前存在，判断clientid是否一致，如果不一致更新为最新的channel
  25.             if (!infoOld.getClientId().equals(infoNew.getClientId())) { 
  26.                 log.error("[BUG] consumer channel exist in broker, but clientId not equal. GROUP: {} OLD: {} NEW: {} ", this.groupName, infoOld.toString(), infoNew.toString());
  27.                 this.channelInfoTable.put(infoNew.getChannel(), infoNew);
  28.             }
  29.         }
  30.         this.lastUpdateTimestamp = System.currentTimeMillis();
  31.         infoOld.setLastUpdateTimestamp(this.lastUpdateTimestamp);
  32.         return updated;
  33.     }
  34. 复制代码

6.2.2 主题信息订阅变更

updateSubscription方法中，主要判断了消费的主题订阅信息是否发生了变化，subscriptionTable中记录了之前记录的订阅信息：

  1. 判断是否有新增的主题订阅信息，主要是通过subscriptionTable是否存在某个主题进行判断的：

    • 如果不存在，表示之前没有订阅过某个主题的信息，将其加入到subscriptionTable中，并将变更状态置为true，表示主题订阅信息有变化；
    • 如果subscriptionTable中存在某个主题的订阅信息，表示之前就已订阅，将其更新为最新的，但是变更状态不发生变化；
  2. 判断是否有删除的主题，主要是通过subscriptionTable和subList的对比进行判断的，如果有删除的主题，将变更状态置为true；

如果消费者订阅的主题发生了变化，比如有新增加的主题或者删除了某个主题的订阅，会被判断为主题订阅信息发生了变化。

  1. public class ConsumerGroupInfo {
  2.     // 记录了订阅的主题信息，key为topic，value为订阅信息
  3.     private final ConcurrentMap<String/* Topic */, SubscriptionData> subscriptionTable =
  4.         new ConcurrentHashMap<String, SubscriptionData>();   
  7.     public boolean updateSubscription(final Set<SubscriptionData> subList) {
  8.         boolean updated = false;
  9.         // 遍历订阅的主题信息
  10.         for (SubscriptionData sub : subList) {
  11.             //根据主题获取订阅信息
  12.             SubscriptionData old = this.subscriptionTable.get(sub.getTopic());
  13.             // 如果获取为空
  14.             if (old == null) {
  15.                 // 加入到subscriptionTable
  16.                 SubscriptionData prev = this.subscriptionTable.putIfAbsent(sub.getTopic(), sub);
  17.                 if (null == prev) {
  18.                     updated = true; // 变更状态置为true
  19.                     log.info("subscription changed, add new topic, group: {} {}", this.groupName, sub.toString());
  20.                 }
  21.             } else if (sub.getSubVersion() > old.getSubVersion()) { // 如果版本发生了变化
  22.                 if (this.consumeType == ConsumeType.CONSUME_PASSIVELY) {
  23.                     log.info("subscription changed, group: {} OLD: {} NEW: {}", this.groupName, old.toString(), sub.toString());
  24.                 }
  25.                 // 更新为最新的订阅信息
  26.                 this.subscriptionTable.put(sub.getTopic(), sub);
  27.             }
  28.         }
  29.         Iterator<Entry<String, SubscriptionData>> it = this.subscriptionTable.entrySet().iterator();
  30.         // 进行遍历，这一步主要是判断有没有取消订阅的主题
  31.         while (it.hasNext()) {
  32.             Entry<String, SubscriptionData> next = it.next();
  33.             String oldTopic = next.getKey();
  34.             boolean exist = false;
  35.             // 遍历最新的订阅信息
  36.             for (SubscriptionData sub : subList) {
  37.                 // 如果在旧的订阅信息中存在就终止，继续判断下一个主题
  38.                 if (sub.getTopic().equals(oldTopic)) {
  39.                     exist = true;
  40.                     break;
  41.                 }
  42.             }
  43.             // 走到这里，表示有取消订阅的主题
  44.             if (!exist) {
  45.                 log.warn("subscription changed, group: {} remove topic {} {}",this.groupName, oldTopic, next.getValue().toString());
  46.                 // 进行删除
  47.                 it.remove();
  48.                 // 变更状态置为true
  49.                 updated = true;
  50.             }
  51.         }
  52.         this.lastUpdateTimestamp = System.currentTimeMillis();
  53.         return updated;
  54.     }
  55. }
  56. 复制代码

6.2.3 变更请求发送

上面讲解了两种被判定为消费者发生变化的情况，被判定为变化之后，会触调用DefaultConsumerIdsChangeListener中的handle方法触发变更事件，在方法中传入了消费者组下的所有消费者的channel对象，会发送变更请求通知该消费者组下的所有消费者，进行负载均衡。

DefaultConsumerIdsChangeListener中处理变更事件时，会对消费组下的所有消费者遍历，调用notifyConsumerIdsChanged方法向每一个消费者发送变更请求:

  1. public class ConsumerManager {
  3.     public boolean registerConsumer(final String group, final ClientChannelInfo clientChannelInfo,
  4.         ConsumeType consumeType, MessageModel messageModel, ConsumeFromWhere consumeFromWhere,
  5.         final Set<SubscriptionData> subList, boolean isNotifyConsumerIdsChangedEnable) {
  6.         // ...
  7.         // 更新Channel
  8.         boolean r1 =
  9.             consumerGroupInfo.updateChannel(clientChannelInfo, consumeType, messageModel,
  10.                 consumeFromWhere);
  11.         // 更新订阅信息
  12.         boolean r2 = consumerGroupInfo.updateSubscription(subList);
  13.         // 如果有变更
  14.         if (r1 || r2) {
  15.             if (isNotifyConsumerIdsChangedEnable) {
  16.                 // 触发变更事件，consumerGroupInfo中存储了该消费者组下的所有消费者的channel
  17.                 this.consumerIdsChangeListener.handle(ConsumerGroupEvent.CHANGE, group, consumerGroupInfo.getAllChannel());
  18.             }
  19.         }
  20.         // ...
  21.     }
  22. }
  24. // DefaultConsumerIdsChangeListener
  25. public class DefaultConsumerIdsChangeListener implements ConsumerIdsChangeListener {
  26.     @Override
  27.     public void handle(ConsumerGroupEvent event, String group, Object... args) {
  28.         if (event == null) {
  29.             return;
  30.         }
  31.         switch (event) {
  32.             case CHANGE:// 如果是消费者变更事件
  33.                 case CHANGE:
  34.                 if (args == null || args.length < 1) {
  35.                     return;
  36.                 }
  37.                 // 获取所有的消费者对应的channel
  38.                 List<Channel> channels = (List<Channel>) args[0];
  39.                 if (channels != null && brokerController.getBrokerConfig().isNotifyConsumerIdsChangedEnable()) {
  40.                     for (Channel chl : channels) {
  41.                         // 向每一个消费者发送变更请求
  42.                         this.brokerController.getBroker2Client().notifyConsumerIdsChanged(chl, group);
  43.                     }
  44.                 }
  45.                 break;
  46.              // ...
  47.         }
  48.     }
  49. }
  50. 复制代码

请求的发送是在Broker2Client的notifyConsumerIdsChanged方法中实现的，可以看到会创建NOTIFY_CONSUMER_IDS_CHANGED请求并发送：

  1. public class Broker2Client {
  2.     public void notifyConsumerIdsChanged(
  3.         final Channel channel,
  4.         final String consumerGroup) {
  5.         if (null == consumerGroup) {
  6.             log.error("notifyConsumerIdsChanged consumerGroup is null");
  7.             return;
  8.         }
  9.         NotifyConsumerIdsChangedRequestHeader requestHeader = new NotifyConsumerIdsChangedRequestHeader();
  10.         requestHeader.setConsumerGroup(consumerGroup);
  11.         // 创建变更通知请求
  12.         RemotingCommand request =
  13.             RemotingCommand.createRequestCommand(RequestCode.NOTIFY_CONSUMER_IDS_CHANGED, requestHeader);
  14.         try {
  15.             // 发送请求
  16.             this.brokerController.getRemotingServer().invokeOneway(channel, request, 10);
  17.         } catch (Exception e) {
  18.             log.error("notifyConsumerIdsChanged exception. group={}, error={}", consumerGroup, e.toString());
  19.         }
  20.     }
  21. }
  22. 复制代码

6.2.4 变更通知请求处理

消费者对Broker发送的NOTIFY_CONSUMER_IDS_CHANGED的请求处理在ClientRemotingProcessor的processRequest方法中，它会调用notifyConsumerIdsChanged方法进行处理，在notifyConsumerIdsChanged方法中可以看到触发了一次负载均衡：

  1. public class ClientRemotingProcessor extends AsyncNettyRequestProcessor implements NettyRequestProcessor {
  3.     @Override
  4.     public RemotingCommand processRequest(ChannelHandlerContext ctx,
  5.         RemotingCommand request) throws RemotingCommandException {
  6.         switch (request.getCode()) {
  7.             case RequestCode.CHECK_TRANSACTION_STATE:
  8.                 return this.checkTransactionState(ctx, request);
  9.             case RequestCode.NOTIFY_CONSUMER_IDS_CHANGED: // NOTIFY_CONSUMER_IDS_CHANGED请求处理
  10.                 // 处理变更请求
  11.                 return this.notifyConsumerIdsChanged(ctx, request); 
  12.             // ...
  13.             default:
  14.                 break;
  15.         }
  16.         return null;
  17.     }
  19.     public RemotingCommand notifyConsumerIdsChanged(ChannelHandlerContext ctx,
  20.         RemotingCommand request) throws RemotingCommandException {
  21.         try {
  22.             // ...
  23.             // 触发负载均衡
  24.             this.mqClientFactory.rebalanceImmediately();
  25.         } catch (Exception e) {
  26.             log.error("notifyConsumerIdsChanged exception", RemotingHelper.exceptionSimpleDesc(e));
  27.         }
  28.         return null;
  29.     }
  30. }
  31. 复制代码

6.3 消费者停止时触发

消费者在停止时，需要将当前消费者负责的消息队列分配给其他消费者进行消费，所以在shutdown方法中会调用MQClientInstance的unregisterConsumer方法取消注册：

  1. public class DefaultMQPushConsumerImpl implements MQConsumerInner {
  2.    public synchronized void shutdown(long awaitTerminateMillis) {
  3.         switch (this.serviceState) {
  4.             case CREATE_JUST:
  5.                 break;
  6.             case RUNNING:
  7.                 this.consumeMessageService.shutdown(awaitTerminateMillis);
  8.                 this.persistConsumerOffset();
  9.                 // 取消注册
  10.                 this.mQClientFactory.unregisterConsumer(this.defaultMQPushConsumer.getConsumerGroup());
  11.                 this.mQClientFactory.shutdown();
  12.                 // ...
  13.                 break;
  14.             case SHUTDOWN_ALREADY:
  15.                 break;
  16.             default:
  17.                 break;
  18.         }
  19.     }
  20. }
  21. 复制代码

在unregisterConsumer方法中，又调用了unregisterClient方法取消注册，与注册消费者的逻辑相似，它会向所有的Broker发送取消注册的请求：

  1. public class MQClientInstance {
  2.     public synchronized void unregisterConsumer(final String group) {
  3.         this.consumerTable.remove(group);
  4.         // 取消注册
  5.         this.unregisterClient(null, group);
  6.     }
  8.     private void unregisterClient(final String producerGroup, final String consumerGroup) {
  9.         // 获取所有的Broker
  10.         Iterator<Entry<String, HashMap<Long, String>>> it = this.brokerAddrTable.entrySet().iterator();
  11.         // 进行遍历
  12.         while (it.hasNext()) {
  13.             // ...
  14.             if (oneTable != null) {
  15.                 for (Map.Entry<Long, String> entry1 : oneTable.entrySet()) {
  16.                     String addr = entry1.getValue();
  17.                     if (addr != null) {
  18.                         try {
  19.                             // 发送取消注册请求
  20.                             this.mQClientAPIImpl.unregisterClient(addr, this.clientId, producerGroup, consumerGroup, clientConfig.getMqClientApiTimeout());
  21.                             // ...
  22.                         } // ...
  23.                     }
  24.                 }
  25.             }
  26.         }
  27.     }
  28. }
  29. 复制代码

取消注册请求的发送是在MQClientAPIImpl中unregisterClient方法实现的，可以看到构建了UNREGISTER_CLIENT请求并发送：

  1. public class MQClientAPIImpl { 
  2.     public void unregisterClient(final String addr, final String clientID, final String producerGroup,  final String consumerGroup, final long timeoutMillis) throws RemotingException, MQBrokerException, InterruptedException {
  3.         // ...
  4.         requestHeader.setConsumerGroup(consumerGroup);
  5.         // 构建UNREGISTER_CLIENT请求
  6.         RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.UNREGISTER_CLIENT, requestHeader);
  7.         // 发送请求
  8.         RemotingCommand response = this.remotingClient.invokeSync(addr, request, timeoutMillis);
  9.         // ...
  10.     }
  11. }
  12. 复制代码

与注册消费者的请求处理一样，Broker对UNREGISTER_CLIENT的请求同样是在ClientManageProcessor的processRequest中处理的，对于UNREGISTER_CLIENT请求是调用unregisterClient方法处理的，里面又调用了ConsumerManager的unregisterConsumer方法进行取消注册：

  1. public class ClientManageProcessor extends AsyncNettyRequestProcessor implements NettyRequestProcessor {
  2.     @Override
  3.     public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request)
  4.         throws RemotingCommandException {
  5.         switch (request.getCode()) {
  6.             case RequestCode.HEART_BEAT: // 处理心跳请求
  7.                 return this.heartBeat(ctx, request);
  8.             case RequestCode.UNREGISTER_CLIENT: // 取消注册请求
  9.                 return this.unregisterClient(ctx, request);
  10.             case RequestCode.CHECK_CLIENT_CONFIG:
  11.                 return this.checkClientConfig(ctx, request);
  12.             default:
  13.                 break;
  14.         }
  15.         return null;
  16.     }
  18.     public RemotingCommand unregisterClient(ChannelHandlerContext ctx, RemotingCommand request)
  19.         throws RemotingCommandException {
  20.         // ...
  21.         {
  22.             final String group = requestHeader.getConsumerGroup();
  23.             if (group != null) {
  24.                 // ...
  25.                 // 取消消费者的注册
  26.                 this.brokerController.getConsumerManager().unregisterConsumer(group, clientChannelInfo, isNotifyConsumerIdsChangedEnable);
  27.             }
  28.         }
  29.         response.setCode(ResponseCode.SUCCESS);
  30.         response.setRemark(null);
  31.         return response;
  32.     }
  33. }
  34. 复制代码

在ConsumerManager的unregisterConsumer方法中，可以看到触发了取消注册事件，之后如果开启了允许通知变更，会触发变更事件，变更事件在上面已经讲解过，它会通知消费者组下的所有消费者进行一次负载均衡：

  1. public class ConsumerManager {    
  2.     public void unregisterConsumer(final String group, final ClientChannelInfo clientChannelInfo,
  3.         boolean isNotifyConsumerIdsChangedEnable) {
  4.         ConsumerGroupInfo consumerGroupInfo = this.consumerTable.get(group);
  5.         if (null != consumerGroupInfo) {
  6.             consumerGroupInfo.unregisterChannel(clientChannelInfo);
  7.             if (consumerGroupInfo.getChannelInfoTable().isEmpty()) {
  8.                 ConsumerGroupInfo remove = this.consumerTable.remove(group);
  9.                 if (remove != null) {
  10.                     // 触发取消注册事件
  11.                     this.consumerIdsChangeListener.handle(ConsumerGroupEvent.UNREGISTER, group);
  12.                 }
  13.             }
  14.             // 触发消费者变更事件
  15.             if (isNotifyConsumerIdsChangedEnable) {
  16.                 this.consumerIdsChangeListener.handle(ConsumerGroupEvent.CHANGE, group, consumerGroupInfo.getAllChannel());
  17.             }
  18.         }
  19.     }
  20. }
  21. 复制代码

6.4 消费者定时触发

在RebalanceService的run方法中，可以看到设置了等待时间，默认是20s，所以消费者本身也会定时执行负载均衡：

  1. public class RebalanceService extends ServiceThread {
  2.     private static long waitInterval = Long.parseLong(System.getProperty("rocketmq.client.rebalance.waitInterval", "20000"));
  4.     @Override
  5.     public void run() {
  6.         log.info(this.getServiceName() + " service started");
  7.         while (!this.isStopped()) {
  8.             this.waitForRunning(waitInterval); // 等待
  9.             // 负载均衡
  10.             this.mqClientFactory.doRebalance();
  11.         }
  12.         log.info(this.getServiceName() + " service end");
  13.     }
  14. }
  15. 复制代码

7.总结

本文主要是从源码角度分析了RocketMQ的重平衡过程，也分析了产生重复消费的原因。简单总结下：

  1. queue的分配算法
  2. 重平衡的代码分析
  3. 重复消费产生的原因
  4. Rebalance的触发时机

作者：hsfxuebao
链接：https://juejin.cn/post/7220380791510040635

发表评论

表情：
评论列表 （有 0 条评论，197人围观）

还没有评论，来说两句吧...

相关阅读

    相关 Dubbo源码 - 负载均衡

    前言 负载均衡，英文名称为Load Balance，其含义就是指将负载（工作任务）进行平衡、分摊到多个操作单元上进行运行。 例如：在Dubbo中，同一个服务有多个服务提

    女爷i女爷i/ 0 赞/ 177 阅读