阿里二面：RocketMQ 集群 Broker 挂了，会造成什么影响？

女爷i 2023-10-01 23:54 9阅读 0赞

大家好，今天分享 RocketMQ 的 Broker 挂了，会带来什么影响。

**面试官**：你好，如果 RocketMQ 集群中的一个 Broker 挂了，会造成什么影响呢？

**我**：Broker 挂了，首先会导致 Producer 发送消息失败。对于普通消息，Producer 同步发送的情况下会有重试机制，重试时把消息发送到其他 Broker。如下图，Broker1 宕机了，把消息发送到了 Broker2：

![0ffea22b883fdc77946fb3b1b12525fb.png][]

发送消息的逻辑其实是是一个循环，发送失败后会不断尝试重新发送，代码如下：

int timesTotal = communicationMode == CommunicationMode.SYNC ? 1 + this.defaultMQProducer.getRetryTimesWhenSendFailed() : 1;
    for (; times < timesTotal; times++) {
     String lastBrokerName = null == mq ? null : mq.getBrokerName();
     MessageQueue mqSelected = this.selectOneMessageQueue(topicPublishInfo, lastBrokerName);
     if (mqSelected != null) {
      mq = mqSelected;
      try {
       sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, topicPublishInfo, timeout - costTime);
       switch (communicationMode) {
        case ASYNC:
         return null;
        case ONEWAY:
         return null;
        case SYNC:
         if (sendResult.getSendStatus() != SendStatus.SEND_OK) {
             //如果发送失败了，这里会进行重试
          if (this.defaultMQProducer.isRetryAnotherBrokerWhenNotStoreOK()) {
           continue;
          }
         }
    
         return sendResult;
        default:
         break;
       }
      } catch (RemotingException e) {
      }//省略其他 catch
     } else {
      break;
     }                                                                                                                                                                                                                                                                                    
    }

对于异步发送和单边消息是不会重试的，因此对于异步和单边消息，就只能发送失败了。而对于同步消息，可以通过重试的方式发送到其他的 Broker 上。

**面试官**：在同步的情况下，Producer 重试时怎么保证不把消息发送到挂掉的 Broker 上呢？

**我**：Producer 默认采用 round-robin 的方式，重试前会记录上一次发送消息的 Broker，然后选择下一个 Broker。代码如下：

//lastBrokerName 记录了上一次发送的 Broker Name
    public MessageQueue selectOneMessageQueue(final String lastBrokerName) {
     if (lastBrokerName == null) {
      return selectOneMessageQueue();
     } else {
      for (int i = 0; i < this.messageQueueList.size(); i++) {
       int index = this.sendWhichQueue.incrementAndGet();
       int pos = Math.abs(index) % this.messageQueueList.size();
       if (pos < 0)
        pos = 0;
       MessageQueue mq = this.messageQueueList.get(pos);
       //Broker Name 不等于上次的，才会返回
       if (!mq.getBrokerName().equals(lastBrokerName)) {
        return mq;
       }
      }
      return selectOneMessageQueue();
     }
    }

**面试官**：在大流量的场景下，可能会有大量消费发送到失败的 Broker，这样导致大量的消息需要重试，对性能影响会很大，有什么解决方法吗？

**我**：RocketMQ 有延迟隔离策略，如果发送某一个 Broker 失败了，会将其隔离，优先选择正常的 Broker 发送消息。**需要注意的是，这个策略默认是不开启的。**

**面试官**：怎么开启延迟隔离策略呢？

**我**：需要在初始化 Producer 的时候定义，见下面代码第二行：

DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
    producer.setSendLatencyFaultEnable(true);
    producer.start();

开启之后，发送消息时会记录发送消息花费的时间（下面 latencyMax 变量），超过一定时间，这个 Broker 就会在一段时间内不允许发送（下面 notAvailableDuration 变量）。

private long[] latencyMax = {50L, 100L, 550L, 1000L, 2000L, 3000L, 15000L};
    private long[] notAvailableDuration = {0L, 0L, 30000L, 60000L, 120000L, 180000L, 600000L};

具体逻辑可以参考类 MQFaultStrategy。

**面试官**：**刚刚聊的是对普通消息的影响，那对顺序消息有什么影响呢？**

**我**：对于**全局顺序消息**，如果设置了所有消息要发送到同一个 Broker 的同一个 MessageQueue 中的情况，恰好是这个 Broker 挂了，那就只能等 Broker 重启后再发送了。而对于**局部顺序消息**，比如同一个订单相关的消息要发送到同一个 Broker 的同一个 MessageQueue 中的情况，如果这个 Broker 挂了，那 MessageQueueSelector 会选择其他 Broker 上的 MessageQueue 进行发送，这会影响当前这笔订单消费的顺序性。而其他订单可以被 Producer 发送到其他的队列中，不受影响。如下图：

![1739d703dfb10b9141e0431e7e18203c.png][]

Broker1 挂之前，Order1 的消息发送到了 Broker1，Broker1 挂之后，Order1 的消息被发送到了 Broker2。在 Broker1 恢复前，消费者只能消费 Broker2 上拉取 Order1 的消息，Broker1 恢复后消费者线程再从 Broker1 拉取，因此 Order1 的消息产生乱序。**这里假设没有从节点**。

**面试官**：Broker 挂了，对 消费者有影响吗？

**我**：如果 Broker 没有设置主从集群，消费者会继续从挂掉的 Broker 上拉取，这会导致拉取失败，直到 NameServer 更新了 Broker 列表。

**面试官**：NameServer 什么时候会更新 Broker 列表呢？

**我**：NameServer 会有每 10s 一次的定时任务检查 Broker 是否下线了，如果 120s 内有没有收到 Broker 心跳，则关闭 channel，把 Broker 信息从本地缓存移除。消费者则默认每隔 30s 向 NameServer 拉取路由信息来刷新本地缓存的 Broker 列表。也就是说可能会有最多 150s 的时间消费者拉取消息失败。如下图：

![d65751704bf424220af55034888348bb.png][]

**面试官**：如果 Broker 集群配置了从节点，还会有上面的影响吗？

**我**：如果有从节点，在 Broker 主节点恢复前，生产者是不能往从节点发送消息的，但是消费者可以去从节点拉取消息。

**面试官**：消费者什么时候会去 Broker 从节点拉取消息呢？

**我**：Broker 挂了以后，消费组会通过向 Name Server 拉取订阅关系来更新本地缓存的 Broker 列表，因为主节点已经不在列表中了，所以会从从节点列表中选择一个 Broker 进项消息拉取。

**面试官**：如果主节点没有挂，消费者会去从节点拉取消息吗？

**我**：在主节点系统压力较大的时候，消费者也会去从节点拉取消息。可以参考下面的代码：

//DefaultMessageStore 类
    //maxOffsetPy:最大物理偏移量
    //maxPhyOffsetPulling:这次消息拉取的最大偏移量
    //diff:还没有被拉取的消息总长度
    long diff = maxOffsetPy - maxPhyOffsetPulling;
    //TOTAL_PHYSICAL_MEMORY_SIZE:系统总的物理内存大小
    //getAccessMessageInMemoryMaxRatio 默认是 40
    long memory = (long) (StoreUtil.TOTAL_PHYSICAL_MEMORY_SIZE
     * (this.messageStoreConfig.getAccessMessageInMemoryMaxRatio() / 100.0));
    getResult.setSuggestPullingFromSlave(diff > memory);

从上面的代码可以看出，当未处理的消息超出物理内存 40% 时就会去从节点拉取。**需要注意两点：**

1.  需要设置 slaveReadEnable 参数为 true，才能去从节点读取数据；
2.  需要配置 whichBrokerWhenConsumeSlowly 参数来决定从哪个从 brokerId 读取。参考下面这段代码：

if (this.brokerController.getBrokerConfig().isSlaveReadEnable()) {
     // consume too slow ,redirect to another machine
     if (getMessageResult.isSuggestPullingFromSlave()) {
         //这里配置从哪个从节点拉取
      responseHeader.setSuggestWhichBrokerId(subscriptionGroupConfig.getWhichBrokerWhenConsumeSlowly());
     }
     //...
    }

1.  brokerId 默认是 0，也就是主节点，如果主节点挂了并且长期启动失败，这个参数也是需要改成可以长期拉取的一个从节点。

**面试官**：Broker 主节点挂了，如果成功从节点拉取消息，可能会重复消费吗？

**我**：对于广播模式，消息偏移量是保存在消费者本地的，只要消费者不挂，按照内存中的偏移量去从节点拉取就行了，不会有问题。对于集群模式，消息偏移量保存在 Broker，路径如下：

/${rocketmq.client.localOffsetStoreDir}/.rocketmq_offsets/${clientId}/${groupName}/offsets.json

消费者消费完一批消息后，会向 Broker 发送请求更新 Broker 内存中保存的偏移量，内存中的偏移量会定时（每 5s 一次）更新到上面文件中。如果 Broker 主节点不挂，无论消费者从主节点还是从节点拉取消息，更新偏移量的请求都会发送到主节点，从节点会每隔 10s 从主节点同步偏移量，如下图：

![6f750bed6e6af107ce122392dae9c82b.png][]

代码如下：

//BrokerController 类 handleSlaveSynchronize
    if (role == BrokerRole.SLAVE) {
     slaveSyncFuture = this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
      @Override
      public void run() {
       try {
        BrokerController.this.slaveSynchronize.syncAll();
       }
      }
     }, 1000 * 3, 1000 * 10, TimeUnit.MILLISECONDS);
    }

**也就是说，如果主节点挂了，去从节点拉取消息，可能因为偏移量没有同步到主节点，从节点保存的偏移量不正确。不过只要消费者不宕机，就会根据消费者本地保存的偏移量去拉取，并不会拉取到重复消息。**

**面试官**：如果 Broker 主节点重启了，主节点并不能同步从节点的最新偏移量，那消费者从主节点读取会读到重复消息吗？

**我**：如果主节点重启了，如果消费者会用本地保存的偏移量去主节点拉取消息，主节点会更新本地的偏移量，同时从节点也会去主节点同步偏移量，所以并不会拉取到重复消息。如果消费者也挂了，消费者重启后 Broker 主节点的偏移量还没有被其他消费者更新过，那确实会拉取到重复消息。

**面试官**：恭喜你，通过了。

[0ffea22b883fdc77946fb3b1b12525fb.png]: https://img-blog.csdnimg.cn/img_convert/0ffea22b883fdc77946fb3b1b12525fb.png
[1739d703dfb10b9141e0431e7e18203c.png]: https://img-blog.csdnimg.cn/img_convert/1739d703dfb10b9141e0431e7e18203c.png
[d65751704bf424220af55034888348bb.png]: https://img-blog.csdnimg.cn/img_convert/d65751704bf424220af55034888348bb.png
[6f750bed6e6af107ce122392dae9c82b.png]: https://img-blog.csdnimg.cn/img_convert/6f750bed6e6af107ce122392dae9c82b.png