阿里二面：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() {
public void run() {
try {
BrokerController.this.slaveSynchronize.syncAll();
}
}
}, 1000 * 3, 1000 * 10, TimeUnit.MILLISECONDS);
}
也就是說，如果主節點掛了，去從節點拉取消息，可能因為偏移量沒有同步到主節點，從節點保存的偏移量不正確。不過只要消費者不宕機，就會根據消費者本地保存的偏移量去拉取，并不會拉取到重復消息。
面試官：如果 Broker 主節點重啟了，主節點并不能同步從節點的最新偏移量，那消費者從主節點讀取會讀到重復消息嗎？
我：如果主節點重啟了，如果消費者會用本地保存的偏移量去主節點拉取消息，主節點會更新本地的偏移量，同時從節點也會去主節點同步偏移量，所以并不會拉取到重復消息。如果消費者也掛了，消費者重啟后 Broker 主節點的偏移量還沒有被其他消費者更新過，那確實會拉取到重復消息。
面試官：恭喜你，通過了。
