RocketMQ主从同步一个重要的特征：主从同步不具备主从切换功能，即当主节点宕机后，从不会接管消息发送，但可以提供消息读取。

温馨提示：本文并不会详细分析RocketMQ主从同步的实现细节，如大家对其感兴趣，可以查阅笔者所著的《RocketMQ技术内幕》或查看笔者博文：https://blog.csdn.net/prestigeding/article/details/79600792

2、提出问题

• 主，从服务器都在运行过程中，消息消费者是从主拉取消息还是从从拉取？
• RocketMQ主从同步架构中，如果主服务器宕机，从服务器会接管消息消费，此时消息消费进度如何保持，当主服务器恢复后，消息消费者是从主拉取消息还是从从服务器拉取，主从服务器之间的消息消费进度如何同步？

接下来带着上述问题，一起来探究其实现原理。

3、原理探究

3.1 RocketMQ主从读写分离机制

RocketMQ的主从同步，在默认情况下RocketMQ会优先选择从主服务器进行拉取消息，并不是通常意义的上的读写分离，那什么时候会从拉取呢？

温馨提示：本节同样不会详细整个流程，只会点出其关键点，如果想详细了解消息拉取、消息消费等核心流程，建议大家查阅笔者所著的《RocketMQ技术内幕》。

在RocketMQ中判断是从主拉取，还是从从拉取的核心代码如下：
DefaultMessageStore#getMessage

long diff = maxOffsetPy - maxPhyOffsetPulling;  // @1 long memory = (long) (StoreUtil.TOTAL_PHYSICAL_MEMORY_SIZE                             * (this.messageStoreConfig.getAccessMessageInMemoryMaxRatio() / 100.0));  // @2 getResult.setSuggestPullingFromSlave(diff > memory);   // @3

代码@1：首先介绍一下几个局部变量的含义：

• maxOffsetPy
  当前最大的物理偏移量。返回的偏移量为已存入到操作系统的PageCache中的内容。
• maxPhyOffsetPulling
  本次消息拉取最大物理偏移量，按照消息顺序拉取的基本原则，可以基本预测下次开始拉取的物理偏移量将大于该值，并且就在其附近。
• diff
  maxOffsetPy与maxPhyOffsetPulling之间的间隔，getMessage通常用于消息消费时，即这个间隔可以理解为目前未处理的消息总大小。

代码@2：获取RocketMQ消息存储在PageCache中的总大小，如果当RocketMQ容量超过该阔值，将会将被置换出内存，如果要访问不在PageCache中的消息，则需要从磁盘读取。

• StoreUtil.TOTAL_PHYSICAL_MEMORY_SIZE
  返回当前系统的总物理内存。参数
• accessMessageInMemoryMaxRatio
  设置消息存储在内存中的阀值，默认为40。
  结合代码@2这两个参数的含义，算出RocketMQ消息能映射到内存中最大值为40% * (机器物理内存)。

代码@3：设置下次拉起是否从从拉取标记，触发下次从从服务器拉取的条件为：当前所有可用消息数据(所有commitlog)文件的大小已经超过了其阔值，默认为物理内存的40%。

那GetResult的suggestPullingFromSlave属性在哪里使用呢？

PullMessageProcessor#processRequest

if (getMessageResult.isSuggestPullingFromSlave()) {      // @1 responseHeader.setSuggestWhichBrokerId(subscriptionGroupConfig.getWhichBrokerWhenConsumeSlowly()); } else {        responseHeader.setSuggestWhichBrokerId(MixAll.MASTER_ID); } switch (this.brokerController.getMessageStoreConfig().getBrokerRole()) {      // @2        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()) { // @3             // 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);      }

代码@1：如果从commitlog文件查找消息时，发现消息堆积太多，默认超过物理内存的40%后，会建议从从服务器读取。

代码@2：如果当前服务器的角色为从服务器:并且slaveReadEnable=true，则忽略代码@1设置的值，下次拉取切换为从主拉取。

代码@3：如果slaveReadEnable=true(从允许读)，并且建议从从服务器读取，则从消息消费组建议当消息消费缓慢时建议的拉取brokerId，由订阅组配置属性whichBrokerWhenConsumeSlowly决定；如果消息消费速度正常，则使用订阅组建议的brokerId拉取消息进行消费，默认为主服务器。如果不允许从可读，则固定使用从主拉取。

温馨提示：请注意broker服务参数slaveReadEnable，与订阅组配置信息：whichBrokerWhenConsumeSlowly、brokerId的值，在生产环境中，可以通过updateSubGroup命令动态改变订阅组的配置信息。

如果订阅组的配置保持默认值的话，拉取消息请求发送到从服务器后，下一次消息拉取，无论是否开启slaveRead