如何解决RocketMQ主从同步若干问题

本文将详细说明如何解决RocketMQ主从同步的一些问题。文章内容质量很高，我就分享给大家作为参考。希望大家看完这篇文章后对相关知识有一定的了解。

00-1010主从同步的基本实现过程如下图所示：。

RocketMQ的主从同步机制如下：

首先启动主机并指定端口监视器；

客户端启动，主动连接到主机，建立TCP连接；

客户端每5秒钟从服务器提取一次消息。如果是第一次拉消息，首先获取本地commitlog文件中的最大偏移量，然后用这个偏移量从服务器拉消息。

服务器分析请求并向客户端返回一批数据；

客户端收到一批消息后，将消息写入本地commitlog文件，然后向Master报告拉取进度，并更新下次要拉取的偏移量；

然后重复步骤3；

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

提示：本文不会详细分析RocketMQ主从同步的实现细节。有兴趣的可以参考作者写的《RocketMQ技术内幕》或者查看我的博文：RocketMQ主从同步详解。

00-1010主服务器和从服务器正在运行。消息消费者是从主服务器还是从服务器获取消息？

在RocketMQ主从同步架构中，如果主服务器宕机，从服务器将接管消息消耗。如何维护消息消费进度？当主服务器恢复时，消息使用者将从主服务器还是从服务器拉消息？如何同步主从服务器之间的消息消耗进度？

接下来，结合上述问题，我们将探讨其实现原理。

1、初识主从同步

-

2、提出问题

RockemQ的主从同步。默认情况下，Rocket MQ会优先从主服务器拉消息，这不是通常意义上的读写分离。它什么时候会从？

小贴士：这一节不会详述整个过程，而只是指出它的要点。想了解更多消息拉取、消息消费等核心流程，请参考作者写的《RocketMQ技术内幕》。

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

long diff=maxOffsetPy-maxPhyOffsetPulling；//[@ 1](https://my . oschina . net/u/1198)

longmemory=(long)(StoreUtil。总物理内存大小

*(this . messagestorecoconfig . getaccessmessageinmemorymexratio()/100.0))；//@2

getresult . setimplementpullingfromserver(diffgt；记忆)；//[@ 3](https://my.oschina.net/u/2648711)代码@1:首先介绍几个局部变量的含义：

MaxOffsetPy当前最大物理偏移量。返回的偏移量是存储在操作系统页面缓存中的内容。

maxphyoffset提取此消息的最大物理偏移量。根据消息顺序拉取的基本原理，基本可以预测下一次要拉取的物理偏移量会大于这个值，并且会在它附近。

diff maxOffsetPy和maxPhyOffsetPulling之间的间隔通常用于getMessage消耗，也就是说，这个间隔可以理解为目前未处理消息的总大小。

Code @2:获取存储在PageCache中的RocketMQ消息的总大小。如果火箭MQ容量超过宽值，

将会将被置换出内存，如果要访问不在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命令动态改变订阅组的配置信息。

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

上面的步骤，在消息拉取命令的返回字段中，会将下次建议拉取Broker返回给客户端，根据其值从指定的broker拉取。

消息拉取实现PullAPIWrapper在处理拉取结果时会将服务端建议的brokerId更新到broker拉取缓存表中。

在发起拉取请求之前，首先根据如下代码，选择待拉取消息的Broker。

3.2 消息消费进度同步机制

从上面内容可知，主从同步引入的主要目的就是消息堆积的内容默认超过物理内存的40%，则消息读取则由从服务器来接管，实现消息的读写分离，避免主服务IO抖动严重。那问题来了，主服务器宕机后，从服务器接管消息消费后，那消息消费进度存储在哪里？当主服务器恢复正常后，消息是从主服务器拉取还是从从服务器拉取？主服务器如何得知最新的消息消费进度呢？

RocketMQ消息消费进度管理（集群模式）： 集群模式下消息消费进度存储文件位于服务端${ROCKETMQ_HOME}/store/config/consumerOffset.json。消息消费者从服务器拉取一批消息后提交到消费组特定的线程池中处理消息，当消息消费成功后会向Broker发送ACK消息，告知消费端已成功消费到哪条消息，Broker收到消息消费进度反馈后，首先存储在内存中，然后定时持久化到consumeOffset.json文件中。备注：关于消息消费进度管理更多的实现细节，建议查阅笔者所著的《RocketMQ技术内幕》。

我们先看一下客户端向服务端反馈消息消费进度时如何选择Broker。 因为主服务的brokerId为0，默认情况下当主服务器存活的时候，优先会选择主服务器，只有当主服务器宕机的情况下，才会选择从服务器。

既然集群模式下消息消费进度存储在Broker端，当主服务器正常时，消息消费进度文件存储在主服务器，那提出如下两个问题： 1）消息消费端在主服务器存活的情况下，会优先向主服务器反馈消息消费进度，那从服务器是如何同步消息消费进度的。 2）当主服务器宕机后则消息消费端会向从服务器反馈消息消费进度，此时消息消费进度如何存储，当主服务器恢复正常后，主服务器如何得知最新的消息消费进度。

为了解开上述两个疑问，我们优先来看一下Broker服务器在收到提交消息消费进度反馈命令后的处理逻辑：

客户端定时向Broker端发送更新消息消费进度的请求，其入口为：RemoteBrokerOffsetStore#updateConsumeOffsetToBroker，该方法中一个非常关键的点是：选择broker的逻辑，如下所示：

如果主服务器存活，则选择主服务器，如果主服务器宕机，则选择从服务器。也就是说，不管消息是从主服务器拉取的还是从从服务器拉取的，提交消息消费进度请求，优先选择主服务器。服务端就是接收其偏移量，更新到服务端的内存中，然后定时持久化到${ROCKETMQ_HOME}/store/config/consumerOffset.json。

经过上面的分析，我们来讨论一下这个场景： 消息消费者首先从主服务器拉取消息，并向其提交消息消费进度，如果当主服务器宕机后，从服务器会接管消息拉取服务，此时消息消费进度存储在从服务器，主从服务器的消息消费进度会出现不一致？那当主服务器恢复正常后，两者之间的消息消费进度如何同步？

3.2.1 从服务定时同步主服务器进度

如果Broker角色为从服务器，会通过定时任务调用syncAll，从主服务器定时同步topic路由信息、消息消费进度、延迟队列处理进度、消费组订阅信息。

那问题来了，如果主服务器启动后，从服务器马上从主服务器同步消息消息进度，那岂不是又要重新消费？

其实在绝大部分情况下，就算从服务从主服务器同步了很久之前的消费进度，只要消息者没有重新启动，就不需要重新消费，在这种情况下，RocketMQ提供了两种机制来确保不丢失消息消费进度。

第一种，消息消费者在内存中存在最新的消息消费进度，继续以该进度去服务器拉取消息后，消息处理完后，会定时向Broker服务器反馈消息消费进度，在上面也提到过，在反馈消息消费进度时，会优先选择主服务器，此时主服务器的消息消费进度就立马更新了，从服务器此时只需定时同步主服务器的消息消费进度即可。

第二种是，消息消费者在向主服务器拉取消息时，如果是是主服务器，在处理消息拉取时，也会更新消息消费进度。

3.2.2 主服务器消息拉取时更新消息消费进度

主服务器在处理消息拉取命令时，会触发消息消费进度的更新，其代码入口为：PullMessageProcessor#processRequest

boolean storeOffsetEnable = brokerAllowSuspend;  // @1
storeOffsetEnable = storeOffsetEnable && hasCommitOffsetFlag; 
storeOffsetEnable = storeOffsetEnable
            && this.brokerController.getMessageStoreConfig().getBrokerRole() != BrokerRole.SLAVE;  // @2
if (storeOffsetEnable) {
            this.brokerController.getConsumerOffsetManager().commitOffset(RemotingHelper.parseChannelRemoteAddr(channel),
                requestHeader.getConsumerGroup(), requestHeader.getTopic(), requestHeader.getQueueId(), requestHeader.getCommitOffset());
 }

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

• brokerAllowSuspend：broker是否允许挂起，在消息拉取时，该值默认为true。

• hasCommitOffsetFlag：消息消费者在内存中是否缓存了消息消费进度，如果缓存了，该标记设置为true。 如果Broker的角色为主服务器，并且上面两个变量都为true，则首先使用commitOffset更新消息消费进度。

看到这里，主从同步消息消费进度的相关问题，应该就有了答案了。

上述实现原理的讲解有点枯燥无味，我们先来回答如下几个问题：

1、主，从服务器都在运行过程中，消息消费者是从主拉取消息还是从从拉取？ 答：默认情况下，RocketMQ消息消费者从主服务器拉取，当主服务器积压的消息超过了物理内存的40%，则建议从从服务器拉取。但如果slaveReadEnable为false，表示从服务器不可读，从服务器也不会接管消息拉取。

2、当消息消费者向从服务器拉取消息后，会一直从从服务器拉取？ 答：不是的。分如下情况： 1）如果从服务器的slaveReadEnable设置为false，则下次拉取，从主服务器拉取。 2）如果从服务器允许读取并且从服务器积压的消息未超过其物理内存的30%，下次拉取使用的Broker为订阅组的brokerId指定的Broker服务器，该值默认为0，代表主服务器。 3）如果从服务器允许读取并且从服务器积压的消息超过了其物理内存的30%，下次拉取使用的Broker为订阅组的whichBrokerWhenConsumeSlowly指定的Broker服务器，该值默认为1，代表从服务器。

3、主从服务消息消费进是如何同步的？ 答：消息消费进度的同步时单向的，从服务器开启一个定时任务，定时从主服务器同步消息消费进度；无论消息消费者是从主服务器拉的消息还是从从服务器拉取的消息，在向Broker反馈消息消费进度时，优先向主服务器汇报；消息消费者向主服务器拉取消息时，如果消息消费者内存中存在消息消费进度时，主会尝试跟新消息消费进度。

读写分离的正确使用姿势： 1、主从Broker服务器的slaveReadEnable设置为true。 2、通过updateSubGroup命令更新消息组whichBrokerWhenConsumeSlowly、brokerId，特别是其brokerId不要设置为0，不然从从服务器拉取一次后，下一次拉取就会从主去拉取。

关于如何解决RocketMQ主从同步若干问题就分享到这里了，希望