kafka offset 机制

kafka消息处理类：MessageAndOffset

case class MessageAndOffset(message: Message, offset: Long) {/*** Compute the offset of the next message in the log*/def nextOffset: Long = offset + 1
}

其主构造方法的参数offset：是这条message上一条的offset =》 lastOffset = messageAndOffset.offset();

其主构造方法的参数message：是需要处理的消息，其offset=lastOffset+1 = messageAndOffset.nextOffset()

lastOffset (即MessageAndOffset.offset)是已经消费过的offset数据（可能是无效message的offset）

kafka consumer 消费数据是通过lastOffset来处理消息message (message.offset=lastOffset+1)

通过lastOffset来判断是否有未处理的数据，若有消息pull下来，封装在MessageAndOffset(message,lastOffset)中

kafka offset是从1开始计数，0是没有数据

创建topic时，创建partition文件，topic没有数据，定义其lastOffset=0，文件名是0

partition的文件名称是以存入数据的第一条message的lastOffset做为文件名

//从zookeeper中获取，获得的是订阅者消费过的offset（即是当前message的lastOffset)；
//message对应的真实offset是(lastOffset+1)
Long offset = kafkaOffset.getOffset(leadBroker, i);
//获得是最旧有效message的lastOffset；可能是通过文件名成来获取的（待源码校验）
Long et=      kafkaOffset.getEarliestOffset(consumer,i);
//topic的最新offset，是有效值。
Long ft=      kafkaOffset.getLatestOffset(consumer,i);

代码实现：

for (MessageAndOffset messageAndOffset : fetchResponse.messageSet(topic, partition)) {long lastOffset = messageAndOffset.offset();if (lastOffset < readOffset) {log.warn("Topic:" + topic + " and Partition:" + partition + "Found an  old offset: " + lastOffset + " Expecting: " + readOffset);continue;}//lastOffset是上次已经消费过的message.offset//每个message的真实offset=（lastOffset+1）readOffset = messageAndOffset.nextOffset();ByteBuffer payload = messageAndOffset.message().payload();byte[] bytes = new byte[payload.limit()];payload.get(bytes);JSONObject value = JSONObject.parseObject(new String(bytes, "utf-8"));list.add(value);
}

kafka offset 机制相关推荐

Kafka设计解析（八）- Kafka事务机制与Exactly Once语义实现原理
http://www.infoq.com/cn/articles/kafka-analysis-part-8# 写在前面的话本文所有Kafka原理性的描述除特殊说明外均基于Kafka 1.0.0版本 ...
springboot手动提交kafka offset
转载自 springboot手动提交kafka offset enable.auto.commit参数设置成了false 但是测试发现enable.auto.commit参数设置成了false,kaf ...
kafka分区机制详解
本文来说下SpringBoot整合kafka之kafka分区实战文章目录 kafka分区机制分区个数选择分区写入策略轮询策略随机策略按键保存策略本文小结 kafka分区机制分区机制是k ...
kafka 同步提交异步_详解Kafka设计架构核心——Kafka副本机制详解
所谓的副本机制(Replication),也可以称之为备份机制,通常是指分布式系统在多台网络互联的机器上保存有相同的数据拷贝.副本机制有什么好处呢? 1. 提供数据冗余.即使系统部分组件失效,系统依然 ...
kafka 消费机制
一.本文介绍了kafka的基础概念:topic.partition.broker.consumer.consumer group和producer. Topic 一个Topic代表了一类资源,一种事件 ...
Kafka 心跳机制重复消费
kafka 心跳机制 Kafka是通过心跳机制来控制消费超时,心跳机制对于消费者客户端来说是无感的,它是一个异步线程,当我们启动一个消费者实例时,心跳线程就开始工作了.心跳超时会导致消息重复消费. 在 ...
Kafka详解（五）Kafka副本机制
所谓的副本机制(Replication),也可以称之为备份机制,通常是指分布式系统在多台网络互联的机器上保存有相同的数据拷贝.它的作用主要有以下几点: 提供数据冗余.即使系统部分组件失效,系统依然能够 ...
6张图阐述Kafka心跳机制（时间轮算法的具体运用）
Broker端与客户端的心跳在Kafka中非常的重要,因为一旦在一个心跳过期周期内(默认10s),Broker端的消费组组协调器(GroupCoordinator)会把消费者从消费组中移除,从而触发重 ...
Kafka | Kafka副本机制详解
今天我要和你分享的主题是:Apache Kafka 的副本机制. 所谓的副本机制(Replication),也可以称之为备份机制,通常是指分布式系统在多台网络互联的机器上保存有相同的数据拷贝.副本机制 ...
大数据Hadoop之——Kafka安全机制（Kafka SSL认证实现）
文章目录一.概述 1)SASL认证概述 2)Delegation Token认证概述 3)SSL认证概述(本章实现) 二.各种安全认证机制对比和使用场景三.Kafka SSL认证实现 1)创建ss ...

kafka offset 机制

kafka offset 机制相关推荐

最新文章

热门文章