RocketMQ（十）——Consumer消费进度（Offset）的管理

文章目录

Consumer消费进度（Offset）的管理
- Offset本地管理模式
- Offset远程管理模式
- offset用途
- 重试队列
- offset的同步提交与异步提交

Consumer消费进度（Offset）的管理

消费进度offset是用来记录每个Queue的不同消费者组的消费进度的。根据消费进度记录器的不用，可以分为两种模式：本地模式和远程模式

Offset本地管理模式

当消费模式为 广播消费时，offset使用本地模式存储。因为每条消息会被素有的消费者消费。每个消费者管理自己的消费进度，各个消费者之间不存在消费进度的交集。
consumer在广播模式下offset相关数据以JSON的形式持久化到consumer 本地磁盘文件中，默认文件路径为当前用户主目录下的.rocketmq_offsets/${当前消费者ID}/${消费者组名}/offsets.json

Offset远程管理模式

当消费者模式为集群消费时，offset使用远程模式管理，因为所有Consumer示例对消息采用的是均衡消费，所有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
当消费完一批消息后，Consumer会提交其消费进度offset给Broker，Broker在收到消费进度后会将其更新到双层Map ConsumerOffsetManager及consumerOffset.json文件中，然后向该Consumer进行ACK，而ACK内容中包含三项数据：minOffset(最小offset)、maxOffset（最大offset）、nextBeginOffset（下次消费的起始offset）。

重试队列

当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（十）——Consumer消费进度（Offset）的管理相关推荐

深入理解RocketMQ：Consumer消费消息原理
前言 RocketMQ版本:4.8 Consumer类型:DefaultMQPushConsumer 原理解析 consumer 启动时做了哪些事情? 定时从NameSrv获取最新的Topic+Que ...
Kafka学习笔记 : 消费进度监控 [ 消费者 Lag 或 Consumer Lag ]
所谓滞后程度,就是指消费者当前落后于生产者的程度. Lag 应该算是最最重要的监控指标了.它直接反映了一个消费者的运行情况.一个正常工作的消费者,它的 Lag 值应该很小,甚至是接近于 0 的,这表示 ...
1.12.Flink Kafka-Connector详解、Consumer消费策略设置、动态加载Topic、Consumers Offset 自动提交、Producer、容错等
1.12.Flink Kafka-Connector详解 1.12.1.Kafka Consumer消费策略设置 1.12.2.Kafka Consumer的容错 1.12.3.动态加载Topic 1 ...
Kafka学习(十)--Kafka消费者Consumer消费消息配置实战
一. Kafka消费者Consumer消费消息配置实战配置: public static Properties getProperties() {Properties props = new Pro ...
RocketMQ：Consumer概述及启动流程与消息拉取源码分析
文章目录 Consumer 概述消费者核心类消费者启动流程消息拉取 PullMessageService实现机制 ProcessQueue实现机制消息拉取基本流程客户端发起消息拉取请求消息 ...
源码分析RocketMQ顺序消息消费实现原理
本节目录 1.消息队列负载 2.消息拉取 3.消息顺序消息消费 3.1核心属性与构造函数 3.2 start方法 3.3 submitConsumeRequest 3.4 ConsumeMessage ...
kafka_消费者组消费进度监控实现
对于 Kafka 消费者,最重要的就是监控它们的消费进度,或者说监控它们消费的滞后程度(消费者 Lag 或 Consumer Lag). 所谓滞后程度,就是指消费者当前落后于生产者的程度.Lag 的单 ...
rocketmq中consumer设计与实现
0.结构图 1.类层次图 2.Consumer消息队列分配消息队列分配使用模板方法模式,在ReblanceImpl中定义处理框架,对于变动部分提炼出抽象方法,交给子类来实现是通过Rebalance ...
kafka 维护消费状态跟踪的方法和消费进度的跟踪
kafka 维护消费状态跟踪大部分消息系统在 broker 端的维护消息被消费的记录:一个消息被分发到consumer 后 broker 就马上进行标记或者等待 customer 的通知后进行标记. ...

RocketMQ（十）——Consumer消费进度（Offset）的管理

文章目录

Consumer消费进度（Offset）的管理

Offset本地管理模式

Offset远程管理模式

offset用途

重试队列

offset的同步提交与异步提交

RocketMQ（十）——Consumer消费进度（Offset）的管理相关推荐

最新文章

热门文章