RocketMQ主从如何同步消息消费进度?
前面我也跟大家讲述了 RocketMQ 读写分离的规则,但是你可能会问,主从服务器之间的消费进度是如何保持同步的?下面我来给大家解答一下。
如果消费者消费模式不同,也会有不同的保存方式,消费者端的消息消费进度保存到 OffsetStore 中,他有两个实现类:
org.apache.rocketmq.client.consumer.store.LocalFileOffsetStore // 本地消费进度保存实现 org.apache.rocketmq.client.consumer.store.RemoteBrokerOffsetStore // 远程消费进度保存实现
其中,如果是广播模式消费,消息的消费进度是保存到本地,如果是集群消费模式,消息的消费进度则是保存到 Broker,但无论是保存到本地,还是保存到 Broker,消费者都会在本地留一份缓存,我们暂且看看集群消费模式下,消息消费进度的缓存是如何保存的:
org.apache.rocketmq.client.consumer.store.RemoteBrokerOffsetStore#updateOffset:
public void updateOffset(MessageQueue mq, long offset, boolean increaseOnly) { if (mq != null) { AtomicLong offsetOld = this.offsetTable.get(mq); if (null == offsetOld) { offsetOld = this.offsetTable.putIfAbsent(mq, new AtomicLong(offset)); } if (null != offsetOld) { if (increaseOnly) { MixAll.compareAndIncreaseOnly(offsetOld, offset); } else { offsetOld.set(offset); } } } }
消息者在消费完消息后,会调用以上方法,讲消费进度放入 offsetTable 缓存中,当 Rebalance 负载重新分配生成 PullRequest 对象时,会调用 RemoteBrokerOffsetStore.readOffset 方法从 offsetTable 缓存中取出对应的消费进度缓存值,再将该值放进 PullRequest 对象中,接下来消息拉取时就很将消息消费进度缓存发送到 Broker 端,所以我们继续看 Broker 端的处理逻辑。
之前整理 Broker 启动流程时,发现 Broker 启动时会开启一个定时任务:
org.apache.rocketmq.broker.BrokerController#initialize:
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { @Override public void run() { try { BrokerController.this.slaveSynchronize.syncAll(); } catch (Throwable e) { log.error("ScheduledTask syncAll slave exception", e); } } }, 1000 * 10, 1000 * 60, TimeUnit.MILLISECONDS);
如果 Broker 是从服务器,则会开启以上定时任务。
org.apache.rocketmq.broker.slave.SlaveSynchronize#syncAll:
public void syncAll() { this.syncTopicConfig(); this.syncConsumerOffset(); this.syncDelayOffset(); this.syncSubscriptionGroupConfig(); }
在主服务器没有宕机的情况下,从服务器会定时从主服务器中同步消息消费进度等信息,那现在问题来了,由于这个同步是单方面同步,即只会从服务器同步主服务器,那如果主服务器宕机了之后,消费者切换成从服务器拉取消息进行消费,如果之后主服务器启动了,从服务器在把已经消费过的偏移量同步过来,那岂不是造成同步消费了?
其实消费者取在拉取消息的时候,如果消费者的缓存中存在消费进度,也会向 Broker 更新消息消费进度,所以即使是主服务器挂了,在它重新启动之后,消费者的消费进度没有丢失,依然会更新主服务器的消息消费进度,这样一来,消费端与主服务器只挂了器中一个,并不会导致消息重新被消费,具体代码逻辑如下:
org.apache.rocketmq.broker.processor.PullMessageProcessor#processRequest:
boolean storeOffsetEnable = brokerAllowSuspend; storeOffsetEnable = storeOffsetEnable && hasCommitOffsetFlag; storeOffsetEnable = storeOffsetEnable && this.brokerController.getMessageStoreConfig().getBrokerRole() != BrokerRole.SLAVE; if (storeOffsetEnable) { this.brokerController.getConsumerOffsetManager().commitOffset(RemotingHelper.parseChannelRemoteAddr(channel), requestHeader.getConsumerGroup(), requestHeader.getTopic(), requestHeader.getQueueId(), requestHeader.getCommitOffset()); }
其中 brokerAllowSuspend 表示 broker 是否允许挂起,该值默认为 true,hasCommitOffsetFlag 表示息消费者在内存中是否缓存了消息消费进度,从代码逻辑可看出,如果 Broker 为主服务器,并且 brokerAllowSuspend 和 hasCommitOffsetFlag 都为true,那么就会将消费者消费进度更新到本地。
关注公众号回复关键字「后端」免费领取后端开发大礼包!
低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。
持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。
转载内容版权归作者及来源网站所有,本站原创内容转载请注明来源。
- 上一篇
RocketMQ主从读写分离机制
一般来说,选择主从备份实现高可用的架构中,都会具备读写分离机制,比如 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_...
- 下一篇
Kafka 重平衡机制
当集群中有新成员加入,或者某些主题增加了分区之后,消费者是怎么进行重新分配消费的?这里就涉及到重平衡(Rebalance)的概念,下面我就给大家讲解一下什么是 Kafka 重平衡机制,我尽量做到图文并茂通俗易懂。 重平衡的作用 重平衡跟消费组紧密相关,它保证了消费组成员分配分区可以做到公平分配,也是消费组模型的实现,消费组模型如下: 从图中可以找到消费组模型的几个概念: 同一个消费组,一个分区只能被一个消费者订阅消费,但一个消费者可订阅多个分区,也即是每条消息只会被同一个消费组的某一个消费者消费,确保不会被重复消费; 一个分区可被不同消费组订阅,这里有种特殊情况,加入每个消费组只有一个消费者,这样分区就会广播到所有消费者上,实现广播模式消费。 要想实现以上消费组模型,那么就要实现当外部环境变化时,比如主题新增了分区,消费组有新成员加入等情况,实现动态调整以维持以上模型,那么这个工作就会交给 Kafka 重平衡机制去处理。 Kafka与RocketMQ的重平衡区别 Kafka 重平衡机制的一些实现相比 RocketMQ 还是有些区别的,但最终的目的还是都是一样,就是保证分区(Rocket...
相关文章
文章评论
共有0条评论来说两句吧...
文章二维码
点击排行
推荐阅读
最新文章
- SpringBoot2初体验,简单认识spring boot2并且搭建基础工程
- Linux系统CentOS6、CentOS7手动修改IP地址
- SpringBoot2全家桶,快速入门学习开发网站教程
- CentOS8编译安装MySQL8.0.19
- CentOS7,CentOS8安装Elasticsearch6.8.6
- SpringBoot2整合Redis,开启缓存,提高访问速度
- Red5直播服务器,属于Java语言的直播服务器
- CentOS6,CentOS7官方镜像安装Oracle11G
- Windows10,CentOS7,CentOS8安装Nodejs环境
- Springboot2将连接池hikari替换为druid,体验最强大的数据库连接池