RocketMQ主從讀寫分離機(jī)制的原理是什么

這篇文章將為大家詳細(xì)講解有關(guān)RocketMQ主從讀寫分離機(jī)制的原理是什么，文章內(nèi)容質(zhì)量較高，因此小編分享給大家做個(gè)參考，希望大家閱讀完這篇文章后對(duì)相關(guān)知識(shí)有一定的了解。

創(chuàng)新互聯(lián)主營(yíng)黃陂網(wǎng)站建設(shè)的網(wǎng)絡(luò)公司,主營(yíng)網(wǎng)站建設(shè)方案,app開發(fā)定制,黃陂h5小程序制作搭建,黃陂網(wǎng)站營(yíng)銷推廣歡迎黃陂等地區(qū)企業(yè)咨詢

RocketMQ 的讀寫分離機(jī)制又跟上述描寫的不太一致，RocketMQ 有屬于自己的一套讀寫分離邏輯，它會(huì)判斷主服務(wù)器的消息堆積量來決定消費(fèi)者是否向從服務(wù)器拉取消息消費(fèi)。

決定消費(fèi)者是否向從服務(wù)器拉取消息消費(fèi)的值存在 GetMessageResult 類中：

org.apache.rocketmq.store.GetMessageResult：

private boolean suggestPullingFromSlave = false;

其默認(rèn)值為 false，即默認(rèn)消費(fèi)者不會(huì)消費(fèi)從服務(wù)器，以下邏輯可以改變?cè)撝担?/p>

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 為當(dāng)前最大物理偏移量，maxPhyOffsetPulling 為本次消息拉取最大物理偏移量，他們的差即可表示消息堆積量，TOTAL_PHYSICAL_MEMORY_SIZE 表示當(dāng)前系統(tǒng)物理內(nèi)存，accessMessageInMemoryMaxRatio 的默認(rèn)值為 40，以上邏輯即可算出當(dāng)前消息堆積量是否大于物理內(nèi)存的 40 %，如果大于則將 suggestPullingFromSlave 設(shè)置為 true。

接下來該參數(shù)值會(huì)在消息拉取邏輯里面產(chǎn)生作用：

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);
}

如果發(fā)現(xiàn)主服務(wù)器的消息堆積超過了物理內(nèi)存的 40%，則會(huì)設(shè)置 suggestWhichBrokerId 為從服務(wù)器 broker ID。

這里還會(huì)有個(gè) slaveReadEnable 值來決定是否可以從從服務(wù)器拉取消息：

如果 slaveReadEnable=true，并且堆積量已經(jīng)超過物理內(nèi)存 40%時(shí)，則建議從從服務(wù)器拉取消息，否則還是從主服務(wù)器拉取消息；
如果 slaveReadEnable=false，則消息者只能從主服務(wù)器中拉取消息。

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);
    }
}

當(dāng)消費(fèi)者收到拉取響應(yīng)回來的數(shù)據(jù)后，會(huì)將下次建議拉取的 brokerID 緩存起來。下次拉取消息就會(huì)從 pullFromWhichNodeTable 中取出拉取 brokerId。

關(guān)于RocketMQ主從讀寫分離機(jī)制的原理是什么就分享到這里了，希望以上內(nèi)容可以對(duì)大家有一定的幫助，可以學(xué)到更多知識(shí)。如果覺得文章不錯(cuò)，可以把它分享出去讓更多的人看到。

網(wǎng)頁名稱：RocketMQ主從讀寫分離機(jī)制的原理是什么
分享網(wǎng)址：http://muchs.cn/article10/ipjogo.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供網(wǎng)站排名、企業(yè)網(wǎng)站制作、網(wǎng)站導(dǎo)航、品牌網(wǎng)站設(shè)計(jì)、網(wǎng)站改版、營(yíng)銷型網(wǎng)站建設(shè)

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請(qǐng)盡快告知，我們將會(huì)在第一時(shí)間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng)，如需處理請(qǐng)聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時(shí)需注明來源：創(chuàng)新互聯(lián)

猜你還喜歡下面的內(nèi)容