RocketMQ主從讀寫分離機制

微信公衆號「後端進階」，專一後端技術分享：Java、Golang、WEB框架、分佈式中間件、服務治理等等。

通常來講，選擇主從備份實現高可用的架構中，都會具有讀寫分離機制，好比 MySql 讀寫分離，客戶端能夠向主從服務器讀取數據，但客戶寫數據只能經過主服務器。

RocketMQ 的讀寫分離機制又跟上述描寫的不太一致，RocketMQ 有屬於本身的一套讀寫分離邏輯，它會判斷主服務器的消息堆積量來決定消費者是否向從服務器拉取消息消費。

決定消費者是否向從服務器拉取消息消費的值存在 GetMessageResult 類中：

org.apache.rocketmq.store.GetMessageResult：

private boolean suggestPullingFromSlave = false;
複製代碼

其默認值爲 false，即默認消費者不會消費從服務器，如下邏輯能夠改變該值：

org.apache.rocketmq.store.DefaultMessageStore#getMessage：

long diff = maxOffsetPy - maxPhyOffsetPulling;
long memory = (long) (StoreUtil.TOTAL_PHYSICAL_MEMORY_SIZE
    * (this.messageStoreConfig.getAccessMessageInMemoryMaxRatio() / 100.0));
getResult.setSuggestPullingFromSlave(diff > memory);
複製代碼

其中 maxOffsetPy 爲當前最大物理偏移量，maxPhyOffsetPulling 爲本次消息拉取最大物理偏移量，他們的差便可表示消息堆積量，TOTAL_PHYSICAL_MEMORY_SIZE 表示當前系統物理內存，accessMessageInMemoryMaxRatio 的默認值爲 40，以上邏輯便可算出當前消息堆積量是否大於物理內存的 40 %，若是大於則將 suggestPullingFromSlave 設置爲 true。

接下來該參數值會在消息拉取邏輯裏面產生做用：

org.apache.rocketmq.broker.processor.PullMessageProcessor#processRequest:

if (getMessageResult.isSuggestPullingFromSlave()) {
  responseHeader.setSuggestWhichBrokerId(subscriptionGroupConfig.getWhichBrokerWhenConsumeSlowly());
} else {
  responseHeader.setSuggestWhichBrokerId(MixAll.MASTER_ID);
}

switch (this.brokerController.getMessageStoreConfig().getBrokerRole()) {
  case ASYNC_MASTER:
  case SYNC_MASTER:
    break;
  case SLAVE:
    if (!this.brokerController.getBrokerConfig().isSlaveReadEnable()) {
      response.setCode(ResponseCode.PULL_RETRY_IMMEDIATELY);
      responseHeader.setSuggestWhichBrokerId(MixAll.MASTER_ID);
    }
    break;
}

if (this.brokerController.getBrokerConfig().isSlaveReadEnable()) {
  // consume too slow ,redirect to another machine
  if (getMessageResult.isSuggestPullingFromSlave()) {
    responseHeader.setSuggestWhichBrokerId(subscriptionGroupConfig.getWhichBrokerWhenConsumeSlowly());
  }
  // consume ok
  else {
    responseHeader.setSuggestWhichBrokerId(subscriptionGroupConfig.getBrokerId());
  }
} else {
  responseHeader.setSuggestWhichBrokerId(MixAll.MASTER_ID);
}
複製代碼

若是發現主服務器的消息堆積超過了物理內存的 40%，則會設置 suggestWhichBrokerId 爲從服務器 broker ID。

這裏還會有個 slaveReadEnable 值來決定是否能夠從從服務器拉取消息：

1. 若是 slaveReadEnable=true，而且堆積量已經超過物理內存 40%時，則建議從從服務器拉取消息，不然仍是從主服務器拉取消息；
2. 若是 slaveReadEnable=false，則消息者只能從主服務器中拉取消息。

org.apache.rocketmq.client.impl.consumer.PullAPIWrapper#updatePullFromWhichNode：

public void updatePullFromWhichNode(final MessageQueue mq, final long brokerId) {
    AtomicLong suggest = this.pullFromWhichNodeTable.get(mq);
    if (null == suggest) {
        this.pullFromWhichNodeTable.put(mq, new AtomicLong(brokerId));
    } else {
        suggest.set(brokerId);
    }
}
複製代碼

當消費者收到拉取響應回來的數據後，會將下次建議拉取的 brokerID 緩存起來。下次拉取消息就會從 pullFromWhichNodeTable 中取出拉取 brokerId。

關注公衆號後回覆關鍵字「後端」免費領取後端開發大禮包！