【RocketMQ工作原理】offset管理
这里的offset指的是Consumer的消费进度offset
消费进度offset是用来记录每个Queue的不同消费组的消费进度的。根据消费进度记录器的不同,可以
分为两种模式:本地模式和远程模式。
offset本地管理模式
当消费模式为广播消费时,offset使用本地模式存储。因为每条消息会被所有的消费者消费,每个消费
者管理自己的消费进度,各个消费者之间不存在消费进度的交集。
Consumer在广播消费模式下offset相关数据以json的形式持久化到Consumer本地磁盘文件中,默认文
件路径为当前用户主目录下的** .rocketmq_offsets/clientId/{clientId}/clientId/{group}/Offsets.json ** 。
其中** clientId∗∗为当前消费者id,默认为∗∗ip@DEFAULT∗∗;{clientId} **为当前消费者id,默认为** ip@DEFAULT **;clientId∗∗为当前消费者id,默认为∗∗ip@DEFAULT∗∗;{group}为消费者组名称。
offset远程管理模式
当消费模式为集群消费时,offset使用远程模式管理。因为所有Cosnumer实例对消息采用的是均衡消
费,所有Consumer共享Queue的消费进度。
Consumer在集群消费模式下offset相关数据以json的形式持久化到Broker磁盘文件中,文件路径为当前
用户主目录下的store/config/consumerOffset.json 。
Broker启动时会加载这个文件,并写入到一个双层Map(ConsumerOffsetManager)。外层map的key
为topic@group,value为内层map。内层map的key为queueId,value为offset。当发生Rebalance时,
新的Consumer会从该Map中获取到相应的数据来继续消费。
集群模式下offset采用远程管理模式,主要是为了保证Rebalance机制。
offset用途
消费者是如何从最开始持续消费消息的?消费者要消费的第一条消息的起始位置是用户自己通过
consumer.setConsumeFromWhere()方法指定的。
在Consumer启动后,其要消费的第一条消息的起始位置常用的有三种,这三种位置可以通过枚举类型
常量设置。这个枚举类型为ConsumeFromWhere。
CONSUME_FROM_LAST_OFFSET:从queue的当前最后一条消息开始消费
CONSUME_FROM_FIRST_OFFSET:从queue的第一条消息开始消费
CONSUME_FROM_TIMESTAMP:从指定的具体时间戳位置的消息开始消费。这个具体时间戳
是通过另外一个语句指定的 。
consumer.setConsumeTimestamp(“20210701080000”) yyyyMMddHHmmss
当消费完一批消息后,Consumer会提交其消费进度offset给Broker,Broker在收到消费进度后会将其更
新到那个双层Map(ConsumerOffsetManager)及consumerOffset.json文件中,然后向该Consumer进
行ACK,而ACK内容中包含三项数据:当前消费队列的最小offset(minOffset)、最大
offset(maxOffset)、及下次消费的起始offset(nextBeginOffset)。
重试队列
当rocketMQ对消息的消费出现异常时,会将发生异常的消息的offset提交到Broker中的重试队列。系统
在发生消息消费异常时会为当前的topic@group创建一个重试队列,该队列以%RETRY%开头,到达重
试时间后进行消费重试。
offset的同步提交与异步提交
集群消费模式下,Consumer消费完消息后会向Broker提交消费进度offset,其提交方式分为两种:
- 同步提交: 消费者在消费完一批消息后会向broker提交这些消息的offset,然后等待broker的成功响
应。若在等待超时之前收到了成功响应,则继续读取下一批消息进行消费(从ACK中获取
nextBeginOffset)。若没有收到响应,则会重新提交,直到获取到响应。而在这个等待过程中,消费
者是阻塞的。其严重影响了消费者的吞吐量。 - 异步提交: 消费者在消费完一批消息后向broker提交offset,但无需等待Broker的成功响应,可以继续读取并消费下一批消息。这种方式增加了消费者的吞吐量。但需要注意,broker在收到提交的offset后,还是会向消费者进行响应的。可能还没有收到ACK,此时Consumer会从Broker中直接获取
nextBeginOffset。
【RocketMQ工作原理】offset管理相关推荐
- RocketMQ(14) -- RocketMQ工作原理-- offset管理
分布式消息队列RocketMQ 3.6)offset管理 这里的offset指的是Consumer的消费进度offset,消费进度offset是用来记录每个Queue的不同消费组的消费进度的.根据消费 ...
- rocketMQ —— 02(集群搭建、rocketmq工作原理)
目录标题 一.相关推荐 二.基本架构图: 三.集群模式 1.单Master模式(这种单节点的理论上不叫集群) 2.多Master模式 3.多Master多Slave模式(异步) 4.多Master多S ...
- 交换机的配置、工作原理以及管理MAC地址表
1.网桥和交换机 网桥: (1)主要基于软件 (2)每个网桥实现一个spanning-tree 交换机 (1)主要基于硬件(ASIC) -(2) 每个交换机可以支持多个spanning-tree sw ...
- 分布式消息队列RocketMQ工作原理与应用(一)
第 1 章 RocketMQ概述 一.MQ概述 1 .MQ简介 MQ,Message Queue,是一种提供消息队列服务的中间件,也称为消息中间件,是一套提供了消息生产.存储.消费全过程API的软件系 ...
- 【RocketMQ工作原理】订阅关系的一致性
订阅关系的一致性指的是,同一个消费者组(Group ID相同)下所有Consumer实例所订阅的Topic与 Tag及对消息的处理逻辑必须完全一致.否则,消息消费的逻辑就会混乱,甚至导致消息丢失. 1 ...
- 【RocketMQ工作原理】indexFile
除了通过通常的指定Topic进行消息消费外,RocketMQ还提供了根据key进行消息查询的功能.该查询 是通过store目录中的index子目录中的indexFile进行索引实现的快速查询.当然,这 ...
- 【RocketMQ工作原理】消息的存储
RocketMQ中的消息存储在本地文件系统中,这些相关文件默认在当前用户主目录下的store目录中. abort:该文件在Broker启动后会自动创建,正常关闭Broker,该文件会自动消失.若在没有 ...
- RocketMQ工作原理 高级功能介绍
1.1 消息存储 分布式队列因为有高可靠性的要求,所以数据要进行持久化存储. 消息生成者发送消息 MQ收到消息,将消息进行持久化,在存储中新增一条记录 返回ACK给生产者 MQ push 消息给对应的 ...
- 【RocketMQ工作原理】消息的清理
消息被消费过后会被清理掉吗?不会的. 消息是被顺序存储在commitlog文件的,且消息大小不定长,所以消息的清理是不可能以消息为单位进行清理的,而是以commitlog文件为单位进行清理的.否则会急 ...
最新文章
- SharePoint 2013的100个新功能之网站管理(二)
- 可以分屏的软件_分享一款非常好用,且小巧的分屏软件
- .NET Core 3.0特性初探:C# 8、WPF、Windows Forms、EF Core
- JS键盘事件最全keycode代码
- AngularJS.js: temple
- Redis的5大Value的使用场景
- 后台开发必备的那些Linux命令
- pdf 模版 汉字和数字_PDF怎么添加数字签名?这款编辑器软件很多人都说好用
- 锐起无盘服务器客户机不同步,使用批处理判断锐起无盘客户机是否为超级用户状态...
- 群晖linux怎么进入u盘,超级简单,群晖系统的U盘制作和安装指导,实现家庭云...
- 2020最强图像匹配综述论文-Part1
- win10系统更新服务器出错怎么办,win10系统正式版自动更新出错的解决方法
- 【中文乱码】深入分析 Java Web 中的中文编码问题
- 喜讯|奇点云联合创始人刘莹荣登“2021最值得关注的女性创业者”榜单
- ValueError X has 2 features, but LogisticRegression is expecting 5 features as input
- python色卡识别_用Python帮小姐姐选口红,人人都是李佳琦
- 微软edge如何退出登录_Microsoft Edge:评论
- 解决Spark数据倾斜(Data Skew)的 N 种姿势 与 问题定位
- 数据仓库(二) 数仓理论(重点核心)
- 【kafka异常】使用Spring-kafka遇到的坑
热门文章
- R语言使用caret包对GBM模型参数调优(自定义调优的评估指标,例如ROC指标):抽取预测标签及类概率、抽样ROC的指标并绘制密度图
- R语言PCA主成分分析(Principle Component Analysis)与线性回归结合实战
- R语言描述性统计分析:假设检验
- python秩和检验(Kruskal-Wallis H Test)
- 机器学习常见的挑战有哪些?
- R 循环中将变量作为对象名
- c语言实现定长顺序存储,c语言:定长顺序串的基本操作实实现
- java 庖丁解牛api_Java Restful API Best Practices
- 计算机应用基础 网上教学设计方案,《计算机应用基础》(本)教学设计方案.doc...
- mysql隐藏用户名_系统默认的MySQL用户名消失的解决方法(修正版)