阿里二面：RocketMQ 集群 Broker 掛了，會造成什么影響？
大家好，我是君哥。今天分享 RocketMQ 的 Broker 掛了，會帶來什么影響。
面試官：你好，如果 RocketMQ 集群中的一個 Broker 掛了，會造成什么影響呢？
我：Broker 掛了，首先會導(dǎo)致 Producer 發(fā)送消息失敗。對于普通消息，Producer 同步發(fā)送的情況下會有重試機制，重試時把消息發(fā)送到其他 Broker。如下圖，Broker1 宕機了，把消息發(fā)送到了 Broker2：
發(fā)送消息的邏輯其實是是一個循環(huán)，發(fā)送失敗后會不斷嘗試重新發(fā)送，代碼如下：
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) {
//如果發(fā)送失敗了，這里會進行重試
if (this.defaultMQProducer.isRetryAnotherBrokerWhenNotStoreOK()) {
continue;
}
}
return sendResult;
default:
break;
}
} catch (RemotingException e) {
}//省略其他 catch
} else {
break;
}
}
對于異步發(fā)送和單邊消息是不會重試的，因此對于異步和單邊消息，就只能發(fā)送失敗了。而對于同步消息，可以通過重試的方式發(fā)送到其他的 Broker 上。
面試官：在同步的情況下，Producer 重試時怎么保證不把消息發(fā)送到掛掉的 Broker 上呢？
我：Producer 默認采用 round-robin 的方式，重試前會記錄上一次發(fā)送消息的 Broker，然后選擇下一個 Broker。代碼如下：
//lastBrokerName 記錄了上一次發(fā)送的 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();
}
}
面試官：在大流量的場景下，可能會有大量消費發(fā)送到失敗的 Broker，這樣導(dǎo)致大量的消息需要重試，對性能影響會很大，有什么解決方法嗎？
我：RocketMQ 有延遲隔離策略，如果發(fā)送某一個 Broker 失敗了，會將其隔離，優(yōu)先選擇正常的 Broker 發(fā)送消息。需要注意的是，這個策略默認是不開啟的。
面試官：怎么開啟延遲隔離策略呢？
我：需要在初始化 Producer 的時候定義，見下面代碼第二行：
DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
producer.setSendLatencyFaultEnable(true);
producer.start();
開啟之后，發(fā)送消息時會記錄發(fā)送消息花費的時間（下面 latencyMax 變量），超過一定時間，這個 Broker 就會在一段時間內(nèi)不允許發(fā)送（下面 notAvailableDuration 變量）。
private long[] latencyMax = {50L, 100L, 550L, 1000L, 2000L, 3000L, 15000L};
private long[] notAvailableDuration = {0L, 0L, 30000L, 60000L, 120000L, 180000L, 600000L};
具體邏輯可以參考類 MQFaultStrategy。
面試官：剛剛聊的是對普通消息的影響，那對順序消息有什么影響呢？
我：對于全局順序消息，如果設(shè)置了所有消息要發(fā)送到同一個 Broker 的同一個 MessageQueue 中的情況，恰好是這個 Broker 掛了，那就只能等 Broker 重啟后再發(fā)送了。而對于局部順序消息，比如同一個訂單相關(guān)的消息要發(fā)送到同一個 Broker 的同一個 MessageQueue 中的情況，如果這個 Broker 掛了，那 MessageQueueSelector 會選擇其他 Broker 上的 MessageQueue 進行發(fā)送，這會影響當(dāng)前這筆訂單消費的順序性。而其他訂單可以被 Producer 發(fā)送到其他的隊列中，不受影響。如下圖：
Broker1 掛之前，Order1 的消息發(fā)送到了 Broker1，Broker1 掛之后，Order1 的消息被發(fā)送到了 Broker2。在 Broker1 恢復(fù)前，消費者只能消費 Broker2 上拉取 Order1 的消息，Broker1 恢復(fù)后消費者線程再從 Broker1 拉取，因此 Order1 的消息產(chǎn)生亂序。這里假設(shè)沒有從節(jié)點。
面試官：Broker 掛了，對 消費者有影響嗎？
我：如果 Broker 沒有設(shè)置主從集群，消費者會繼續(xù)從掛掉的 Broker 上拉取，這會導(dǎo)致拉取失敗，直到 NameServer 更新了 Broker 列表。
面試官：NameServer 什么時候會更新 Broker 列表呢？
我：NameServer 會有每 10s 一次的定時任務(wù)檢查 Broker 是否下線了，如果 120s 內(nèi)有沒有收到 Broker 心跳，則關(guān)閉 channel，把 Broker 信息從本地緩存移除。消費者則默認每隔 30s 向 NameServer 拉取路由信息來刷新本地緩存的 Broker 列表。也就是說可能會有最多 150s 的時間消費者拉取消息失敗。如下圖：
面試官：如果 Broker 集群配置了從節(jié)點，還會有上面的影響嗎？
我：如果有從節(jié)點，在 Broker 主節(jié)點恢復(fù)前，生產(chǎn)者是不能往從節(jié)點發(fā)送消息的，但是消費者可以去從節(jié)點拉取消息。
面試官：消費者什么時候會去 Broker 從節(jié)點拉取消息呢？
我：Broker 掛了以后，消費組會通過向 Name Server 拉取訂閱關(guān)系來更新本地緩存的 Broker 列表，因為主節(jié)點已經(jīng)不在列表中了，所以會從從節(jié)點列表中選擇一個 Broker 進項消息拉取。
面試官：如果主節(jié)點沒有掛，消費者會去從節(jié)點拉取消息嗎？
我：在主節(jié)點系統(tǒng)壓力較大的時候，消費者也會去從節(jié)點拉取消息?？梢詤⒖枷旅娴拇a：
//DefaultMessageStore 類
//maxOffsetPy:最大物理偏移量
//maxPhyOffsetPulling:這次消息拉取的最大偏移量
//diff:還沒有被拉取的消息總長度
long diff = maxOffsetPy - maxPhyOffsetPulling;
//TOTAL_PHYSICAL_MEMORY_SIZE:系統(tǒng)總的物理內(nèi)存大小
//getAccessMessageInMemoryMaxRatio 默認是 40
long memory = (long) (StoreUtil.TOTAL_PHYSICAL_MEMORY_SIZE
* (this.messageStoreConfig.getAccessMessageInMemoryMaxRatio() / 100.0));
getResult.setSuggestPullingFromSlave(diff > memory);
從上面的代碼可以看出，當(dāng)未處理的消息超出物理內(nèi)存 40% 時就會去從節(jié)點拉取。需要注意兩點：
- 需要設(shè)置 slaveReadEnable 參數(shù)為 true，才能去從節(jié)點讀取數(shù)據(jù)。
- 需要配置 whichBrokerWhenConsumeSlowly 參數(shù)來決定從哪個從 brokerId 讀取。參考下面這段代碼：
if (this.brokerController.getBrokerConfig().isSlaveReadEnable()) {
// consume too slow ,redirect to another machine
if (getMessageResult.isSuggestPullingFromSlave()) {
//這里配置從哪個從節(jié)點拉取
responseHeader.setSuggestWhichBrokerId(subscriptionGroupConfig.getWhichBrokerWhenConsumeSlowly());
}
//...
}
- brokerId 默認是 0，也就是主節(jié)點，如果主節(jié)點掛了并且長期啟動失敗，這個參數(shù)也是需要改成可以長期拉取的一個從節(jié)點。
面試官：Broker 主節(jié)點掛了，如果成功從節(jié)點拉取消息，可能會重復(fù)消費嗎？
我：對于廣播模式，消息偏移量是保存在消費者本地的，只要消費者不掛，按照內(nèi)存中的偏移量去從節(jié)點拉取就行了，不會有問題。對于集群模式，消息偏移量保存在 Broker，路徑如下：
/${rocketmq.client.localOffsetStoreDir}/.rocketmq_offsets/${clientId}/${groupName}/offsets.json
消費者消費完一批消息后，會向 Broker 發(fā)送請求更新 Broker 內(nèi)存中保存的偏移量，內(nèi)存中的偏移量會定時（每 5s 一次）更新到上面文件中。如果 Broker 主節(jié)點不掛，無論消費者從主節(jié)點還是從節(jié)點拉取消息，更新偏移量的請求都會發(fā)送到主節(jié)點，從節(jié)點會每隔 10s 從主節(jié)點同步偏移量，如下圖：
代碼如下：
//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);
}
也就是說，如果主節(jié)點掛了，去從節(jié)點拉取消息，可能因為偏移量沒有同步到主節(jié)點，從節(jié)點保存的偏移量不正確。不過只要消費者不宕機，就會根據(jù)消費者本地保存的偏移量去拉取，并不會拉取到重復(fù)消息。
面試官：如果 Broker 主節(jié)點重啟了，主節(jié)點并不能同步從節(jié)點的最新偏移量，那消費者從主節(jié)點讀取會讀到重復(fù)消息嗎？
我：如果主節(jié)點重啟了，如果消費者會用本地保存的偏移量去主節(jié)點拉取消息，主節(jié)點會更新本地的偏移量，同時從節(jié)點也會去主節(jié)點同步偏移量，所以并不會拉取到重復(fù)消息。如果消費者也掛了，消費者重啟后 Broker 主節(jié)點的偏移量還沒有被其他消費者更新過，那確實會拉取到重復(fù)消息。
面試官：恭喜你，通過了。
