RocketMQ主从读写分离机制的原理是什么

这篇文章将为大家详细讲解有关RocketMQ主从读写分离机制的原理是什么，文章内容质量较高，因此小编分享给大家做个参考，希望大家阅读完这篇文章后对相关知识有一定的了解。

创新互联主营黄陂网站建设的网络公司,主营网站建设方案,app开发定制,黄陂h5小程序制作搭建,黄陂网站营销推广欢迎黄陂等地区企业咨询

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。

关于RocketMQ主从读写分离机制的原理是什么就分享到这里了，希望以上内容可以对大家有一定的帮助，可以学到更多知识。如果觉得文章不错，可以把它分享出去让更多的人看到。