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

大家好，阿里我是影响君哥。今天分享 RocketMQ 的阿里 Broker 挂了，会带来什么影响。影响

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

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

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

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 发送到其他的队列中，不受影响。如下图：

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 的时间消费者拉取消息失败。如下图：

面试官：如果 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% 时就会去从节点拉取。需要注意两点：

需要设置 slaveReadEnable 参数为 true，才能去从节点读取数据。需要配置 whichBrokerWhenConsumeSlowly 参数来决定从哪个从 brokerId 读取。参考下面这段代码：if (this.brokerController.getBrokerConfig().isSlaveReadEnable()) {

// consume too slow ,redirect to another machine

if (getMessageResult.isSuggestPullingFromSlave()) {

//这里配置从哪个从节点拉取

responseHeader.setSuggestWhichBrokerId(subscriptionGroupConfig.getWhichBrokerWhenConsumeSlowly());

}

//...

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

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

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

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

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

代码如下：

//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 主节点的偏移量还没有被其他消费者更新过，那确实会拉取到重复消息。

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