kafka offset机制（Kafka中的索引机制）

本文目录

• Kafka中的索引机制
• Kafka的Offset、Index(三)
• Kafka Consumer Offset解析
• kafka消费者和offset的关系，以及异常处理问题
• kafka 提交offset
• kafka consumer offset机制

Kafka中的索引机制

在kafka中，每个日志分段文件都对应了两个索引文件—— 偏移量索引文件和时间戳索引文件 （还有其它的诸如事务日志索引文件就不细表了），主要用来 提高查找消息的效率 。

偏移量索引文件用来建立消息偏移量(offset)到物理地址之间的映射关系，方便快速定位消息所在的物理文件位置；时间戳索引文件则根据指定的时间戳(timestamp)来查找对应的偏移量信息。

Kafka 中的索引文件以稀疏索引(sparse index)的方式构造消息的索引，它并不保证每个消息在索引文件中都有对应的索引项。

每当写入一定量 (由 broker 端参数 log.index.interval.bytes 指定，默认值为 4096，即 4KB) 的消息时，偏移量索引文件和时间戳索引文件分别增加一个偏移量索引项和时间戳索引项，增大或减小 log.index.interval.bytes 的值，对应地可以缩小或增加索引项的密度。

稀疏索引通过 MappedByteBuffer 将索引文件映射到内存中，以加快索引的查询速度。

偏移量索引文件中的偏移量是单调递增的，查询指定偏移量时，使用二分查找法来快速定位偏移量的位置，如果指定的偏移量不在索引文件中，则会返回小于指定偏移量的最大偏移量。

时间戳索引文件中的时间戳也保持严格的单调递增，查询指定时间戳时，也根据二分查找法来查找不大于该时间戳的最大偏移量，至于要找到对应的物理文件位置还需要根据偏移量索引文件来进行再次定位。

稀疏索引的方式是在磁盘空间、内存空间、查找时间等多方面之间的一个折中。

以偏移量索引文件来做具体分析。偏移量索引项的格式如下图所示。

每个索引项占用 8 个字节，分为两个部分：

(1) relativeOffset : 相对偏移量，表示消息相对于 baseOffset 的偏移量，占用 4 个字节，当前索引文件的文件名即为 baseOffset 的值。

(2) position : 物理地址，也就是消息在日志分段文件中对应的物理位置，占用 4 个字节。

消息的偏移量(offset)占用 8 个字节，也可以称为绝对偏移量。

索引项中没有直接使用绝对偏移量而改为只占用 4 个字节的相对偏移量(relativeOffset = offset - baseOffset)，这样可以减小索引文件占用的空间。

举个例子，一个日志分段的 baseOffset 为 32，那么其文件名就是 00000000000000000032.log，offset 为 35 的消息在索引文件中的 relativeOffset 的值为 35-32=3。

如果我们要查找偏移量为 23 的消息，那么应该怎么做呢?首先通过二分法在偏移量索引文件中找到不大于 23 的最大索引项，即，然后从日志分段文件中的物理位置 656 开始顺序查找偏移量为 23 的消息。

以上是最简单的一种情况。参考上图，如果要查找偏移量为 268 的消息，那么应该怎么办呢?

首先肯定是定位到baseOffset为251的日志分段，然后计算相对偏移量relativeOffset = 268 - 251 = 17，之后再在对应的索引文件中找到不大于 17 的索引项，最后根据索引项中的 position 定位到具体的日志分段文件位置开始查找目标消息。

那么又是如何查找 baseOffset 为 251 的日志分段的呢?

这里并不是顺序查找，而是用了跳跃表的结构。

Kafka 的每个日志对象中使用了 ConcurrentSkipListMap 来保存各个日志分段，每个日志分段的 baseOffset 作为 key，这样可以根据指定偏移量来快速定位到消息所在的日志分段。

在Kafka中要定位一条消息，那么首先根据 offset 从 ConcurrentSkipListMap 中来查找到到对应（baseOffset）日志分段的索引文件，然后读取偏移量索引索引文件，之后使用二分法在偏移量索引文件中找到不大于 offset - baseOffset z的最大索引项，接着再读取日志分段文件并且从日志分段文件中顺序查找relativeOffset对应的消息。

Kafka中通过offset查询消息内容的整个流程我们可以简化成下图：

Kafka中消息的offset可以类比成InnoDB中的主键，前者是通过offset检索出整条Record的数据，后者是通过主键检索出整条Record的数据。

InnoDB中通过主键查询数据内容的整个流程建议简化成下图(下半部分)。

Kafka中通过时间戳索引文件去检索消息的方式可以类比于InnoDB中的辅助索引的检索方式：

前者是通过timestamp去找offset，后者是通过索引去找主键，后面两者的过程就和上面的陈述相同。

Kafka中当有新的索引文件建立的时候ConcurrentSkipListMap才会更新，而不是每次有数据写入时就会更新，这块的维护量基本可以忽略

B+树中数据有插入、更新、删除的时候都需要更新索引，还会引来“页分裂”等相对耗时的操作。Kafka中的索引文件也是顺序追加文件的操作，和B+树比起来工作量要小很多。

说到底还是应用场景不同所决定的。MySQL中需要频繁地执行CRUD的操作，CRUD是MySQL的主要工作内容，而为了支撑这个操作需要使用维护量大很多的B+树去支撑。

Kafka中的消息一般都是顺序写入磁盘，再到从磁盘顺序读出（不深入探讨page cache等），他的主要工作内容就是：写入+读取，很少有检索查询的操作

换句话说， 检索查询只是Kafka的一个辅助功能，不需要为了这个功能而去花费特别太的代价去维护一个高level的索引。

前面也说过，Kafka中的这种方式是在磁盘空间、内存空间、查找时间等多方面之间的一个折中。

Kafka的Offset、Index(三)

Kafka中的每个Partition都由一系列有序的、不可变的消息组成，这些消息被连续的追加到Partition中。Partition中的每个消息都有一个连续的序号，用于Partition唯一标识一条消息。
Offset记录着下一条将要发送给Consumer的消息的序号。
Offset从语义上来看拥有两种： Current Offset 和 Committed Offset 。

Current Offset保存在Consumer客户端中，它表示Consumer希望收到的下一条消息的序号。它仅仅在poll()方法中使用。如，Consumer第一次调用poll()方法后收到了20条消息，那么Current Offset就被设置为20。这样Consumer下一次调用poll()方法时，Kafka就知道应该从序号为21的消息开始读取。这样就能够保证每次Consumer poll消息时，都能够收到不重复的消息。

Committed Offset保存在Broker上（V0.9之后的版本），它表示Consumer已经确认消费过的消息的序号。主要通过commitSync()来操作。举例: Consumer通过poll() 方法收到20条消息后，此时Current Offset就是20，经过一系列的逻辑处理后，并没有调用commitSync()来提交Committed Offset，那么此时Committed Offset依旧是0。

Committed Offset主要用于Consumer Rebalance(再平衡)。在Consumer Rebalance的过程中，一个Partition被分配给了一个Consumer，那么这个Consumer该从什么位置开始消费消息呢？答案就是Committed Offset。另外，如果一个Consumer消费了5条消息（poll并且成功commitSync）之后宕机了，重新启动之后，它仍然能够从第6条消息开始消费，因为Committed Offset已经被Kafka记录为5。

小结一下 ：

在Kafka V0.9前，Committed Offset信息保存在zookeeper的目录中（zookeeper其实并不适合进行大批量的读写操作，尤其是写操作）。在V0.9之后，所有的offset信息都保存在了Broker上的一个名为__consumer_offsets的topic(系统自维护的)中。

auto.offset.reset表示如果Kafka中没有存储对应的offset信息的话（有可能offset信息被删除），消费者从何处开始消费消息。有三个可选值：

分两个场景来说明：
a) Consumer消费了5条消息后宕机了，重启之后它读取到对应的Partition的Committed Offset为5，因此会直接从第6条消息开始读取。此时完全依赖于Committed Offset机制，和auto.offset.reset配置完全无关。
b) 新建了一个新的Group，并添加了一个Consumer，它订阅了一个已经存在的Topic。此时Kafka中还没有这个Consumer相应的Offset信息，因此此时Kafka就会根据auto.offset.reset配置来决定这个Consumer从何处开始消费消息。

在Kafka文件存储中，同一个topic下有多个不同的partition，每个partiton为一个目录，partition的名称规则为：topic名称+有序序号，第一个序号从0开始计，最大的序号为partition数量减1，partition是实际物理上的概念，而topic是逻辑上的概念。

如果就以partition为最小存储单位，我们可以想象当Kafka producer不断发送消息，必然会引起partition文件的无限扩张，这样对于消息文件的维护以及已经被消费的消息的清理带来严重的影响，所以这里以segment为单位又将partition细分。每个partition(目录)相当于一个巨型文件被平均分配到多个大小相等的segment(段)数据文件中（每个segment 文件中消息数量不一定相等）这种特性也方便old segment的删除，即方便已被消费的消息的清理，提高磁盘的利用率。每个partition只需要支持顺序读写就行，segment的文件生命周期由服务端配置参数（log.segment.bytes，log.roll.{ms,hours}等若干参数）决定。

segment文件由两部分组成，分别为“.index”文件和“.log”文件，分别表示为segment索引文件和数据文件。这两个文件的命令规则为：Partition全局的第一个segment从0开始，后续每个segment文件名为上一个segment文件最后一条消息的offset值，数值大小为64位，20位数字字符长度，没有数字用0填充，如下：

以上面的segment文件为例，展示出segment：00000000000000170410的“.index”文件和“.log”文件的对应的关系，如下图：

如上图，“.index”索引文件存储大量的元数据，“.log”数据文件存储大量的消息，索引文件中的元数据指向对应数据文件中message的物理偏移地址。其中以“.index”索引文件中的元数据为例，在“.log”数据文件表示第3个消息，即在全局partition中表示170410+3=170413个消息，该消息的物理偏移地址为348。

那么如何从partition中通过offset查找message呢？以上图为例，读取offset=170418的消息，首先查找segment文件，其中00000000000000000000.index为最开始的文件，第二个文件为00000000000000170410.index（起始偏移为170410+1=170411），而第三个文件为00000000000000239430.index（起始偏移为239430+1=239431），所以这个offset=170418就落到了第二个文件之中。其他后续文件可以依次类推，以其实偏移量命名并排列这些文件，然后根据二分查找法就可以快速定位到具体文件位置。其次根据00000000000000170410.index文件中的定位到00000000000000170410.log文件中的1325的位置进行读取。

———————————————————
坐标帝都，白天上班族，晚上是知识的分享者
如果读完觉得有收获的话，欢迎点赞加关注

Kafka Consumer Offset解析

Kafka __consumer_offsets是一个特殊的存储元数据的Topic
数据格式可以想象成一个 KV 格式的消息，key 就是一个三元组：group.id+topic+分区号，而 value 就是 offset 的值。

查看方式:使用kafka自带的读取类
./bin/kafka-console-consumer.sh --topic __consumer_offsets --partition 01 --bootstrap-server xxx:9092 --formatter "kafka.coordinator.group.GroupMetadataManager$OffsetsMessageFormatter" --from-beginning --max-messages 30

一般情况下， 使用 OffsetsMessageFormatter 打印的格式可以概括为：
"".format(group, topic, partition, offset, metadata, commitTimestamp, expireTimestamp)

数据内容:
::NULL
::NULL
::NULL

另外一种是
::OffsetAndMetadata(offset=19, leaderEpoch=Optional.empty, metadata=, commitTimestamp=1636939024066, expireTimestamp=None)
::OffsetAndMetadata(offset=19, leaderEpoch=Optional.empty, metadata=, commitTimestamp=1636939028621, expireTimestamp=None)
::OffsetAndMetadata(offset=19, leaderEpoch=Optional.empty, metadata=, commitTimestamp=1636939033680, expireTimestamp=None)

还有一种是




分别解释一下:

在 Kafka 中有一个名为“delete-expired-group-metadata”的定时任务来负责清理过期的消费位移，这个定时任务的执行周期由参数 offsets.retention.check.interval.ms 控制，默认值为600000，即10分钟。这和普通的topic的不太一样

还有 metadata，一般情况下它的值要么为 null 要么为空字符串，OffsetsMessageFormatter 会把它展示为 NO_METADATA，否则就按实际值进行展示。

看一下源码里这些类的结构
case class OffsetAndMetadata(offsetMetadata: OffsetMetadata,
commitTimestamp: Long = org.apache.kafka.common.requests.OffsetCommitRequest.DEFAULT_TIMESTAMP,
expireTimestamp: Long = org.apache.kafka.common.requests.OffsetCommitRequest.DEFAULT_TIMESTAMP) {

case class OffsetMetadata(offset: Long, metadata: String = OffsetMetadata.NoMetadata) {
override def toString = "OffsetMetadata"
.format(offset,
if (metadata != null && metadata.length 》 0) metadata else "NO_METADATA")
}

@Deprecated
public static final long DEFAULT_TIMESTAMP = -1L; // for V0, V1

另外0.11.0之后对应的数据格式版本是V2,这个版本的消息相比于v0和v1的版本而言改动很大，同时还参考了Protocol Buffer而引入了变长整型（Varints）和ZigZag编码。

另外:
offset为什么会有墓碑消息?
因为offset本身也会过期清理.受offsets.retention.minutes 这个配置的影响
看下官网介绍
After a consumer group loses all its consumers (i.e. becomes empty) its offsets will be kept for this retention period before getting discarded. For standalone consumers (using manual assignment), offsets will be expired after the time of last commit plus this retention period.
当group里的consumer全部下线后过offsets.retention.minutes 时间后offset就会被删除
val OffsetsRetentionMinutes: Int = 7 * 24 * 60 // 默认7天
默认2.0之前是1天,2.0及以后是7天 这个官方真是..要么就改为2天,结果直接改为7天,改动不可谓不大,而且active的group不会过期

***隐藏网址***

另外active的group无法修改consumer offset?
Usually we do not allow committed offset changes while a group is active because we do not have a mechanism to notify the group of the change.
原因是无法通知到组成员consumer offset的变更

kafka消费者和offset的关系，以及异常处理问题

earliest： 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
latest： 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
none： topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常

简单来说，如果partition里已经有数据，但还没有消费，earliest就会从没消费的起始点来消费，反观latest就不会去消费；如果partition已经有已消费的数据，再放新的数据进去，那么它们都会从新的数据开始消费。

offset会保存在kafka内部，一开始发送数据到kafka的时候就有offset，只是有没有提交而已。而使用spring-kafka时，客户端在监听topic的时候，它有2种提交offset的方式：

1、自动提交，设置enable.auto.commit=true，更新的频率根据参数【auto.commit.interval.ms】来定。这种方式也被称为【at most once】，fetch到消息后就可以更新offset，无论是否消费成功。

2、手动提交，设置enable.auto.commit=false，这种方式称为【at least once】。fetch到消息后，等消费完成再调用方法【consumer.commitSync()】，手动更新offset；如果消费失败，则offset也不会更新，此条消息会被重复消费一次。

***隐藏网址***

因为是批量消费，所以@KafkaListener需要使用list来接收消息，如果使用单个bean会报错。正常不设置异常处理，它会不断循环重复消费这条数据，不像别的地方说有一定数量的重试。

实现接口new BatchErrorHandler自定义属于自己的批量异常处理，但只会到：
public void handle(Exception thrownException, ConsumerRecords《?, ?》 data)

而不到
public void handle(Exception thrownException, ConsumerRecords《?, ?》 data, Consumer《?, ?》 consumer, MessageListenerContainer container)

再定义逻辑自定处理。如果像官网那样seek回开始的offset，也是无限循环，不太了解所以不采用。

实现接口ConsumerAwareListenerErrorHandler，注意区别是有个Listener的，

如果同时存在局部和全局，在@KafkaListener注解中标注了这个局部的异常处理器，会优先使用局部的。

如果发生异常，来的是一批数据，如果头部发生了异常，那么后面的都会略过。按照参考链接中的异常处理，定义一个死信，来接收这些失败的msg，如果异常处理在全局异常处理器中，那么它们都被发送到死信，后续就算数据是正确的，都不会处理，所以 还是建议个人使用try catch来包裹 处理，个人尝试在kafka处理业务远程插入两条数据，第一条错误，第二条正确，try catch中第一条自定发到死信，第二条会正确入库。

参考：
kafka之consumer参数auto.offset.reset

kafka 消费者offset记录位置和方式

消息队列-kafka消费异常问题

Kafka - 异常处理 待试

kafka 提交offset

每次调用poll()方法，它总是返回由生产者写入kafka但还没有被消费者读取过的记录，我们因此可以追溯到哪些记录是被群组里的那个消费者读取的。之前已经讨论过，Kafka不会像其他的JMS队列那样需要得到消费者的确认，这是kafka的一个独特之处。相反，消费者可以使用kafka来追踪消息在分区的位置（偏移量）
那么消费者是如何提交offset的呢？consumer 往一个叫做_consumer_offset的特殊主题发送消息，消息里面包含每个分区的偏移量。如果消费者一直处于运行状态，那么偏移量就没有什么用处。不过，如果消费者发生崩溃或者有新的消费者加入群组，就会触发rebanlance(再均衡)，完成在均衡之后，每个消费者可能分配到新的分区，而不是之前处理的那个，为了能够继续之前的工作，消费者需要读取每个分区最后一次提交的偏移量，然后从偏移量指定的地方继续处理。
Q1 如果提交的偏移量小于客户端处理的最后一个消息的offset，则两者之间的数据就会被重复消费。
Q2 如果提交的偏移量大于客户端处理的最后一个消息的offset,则两者职期间的数据就会丢失。
所以，偏移量的提交对客户端有很大的影响。

最简单的方式就是consumer自动提交offset，如果enable.auto.commit =true，那么每过5s,consumer会自动把poll()方法接收到的最大offset提交上去。提交时间间隔由auto.commit.interval.ms 控制，默认是 5s.与消费者里其他的东西一样，自动提交也是在轮询里进行的。consumer每次在进行查询的时候回检查是否该提交偏移量了，如果是，那么就会提交从上一次轮询返回的偏移量。
不过， 在使用这种渐变的方式之前，需要知道它将会带来怎样的后果。
假设我们使用默认的5s提交时间间隔，在最近一次提交之后的3是，发生了在均衡，在均衡之后，消费者从最后一次提交的offset的位置开始读取消息，这个时候offset已经落后了3s,所以在这3s到达的消息会被重复处理。可以通过修改提交时间来频繁的提交offset，减少可能出现重复消息的时间窗，不过这种情况是无法完全避免的。

处理完当前批次的消息，在轮询更多的消息之前，调用commitSync方法提交当前批次最新的offset
只要没有发生不可恢复的错误，commitSync()会一直尝试直至提交成功，如果提交失败，我们也只能把异常记录到日志里。

提交一个offset，然后继续做其他事情，如果提交失败，错误信息和偏移量会被记录下来。commitAsync和commitSync不同在于，它不会一直重试，是因为有可能在它收到服务器响应之前，可能有一个更大的offset已经提交成功。另外commitAsync支持回调。

本文摘自kafka权威指南

kafka consumer offset机制

kafka消息在分区中是按序一条一条存储的，假如分区中有10条消息，位移就是0-9，

consumer消费了5条消息，那么offset就是5，指向了下一条要消费的记录，consumer

需要向kafka汇报自己的位移数据，因为consumer是能够消费多个分区的，所以offset

的粒度是分区，consumer需要为分配给他的各分区分别提交offset信息。

从用户的角度来说，位移提交分为自动提交和手动提交，在consumer的角度来说，位移

分为同步提交和异步提交。

kafka内部有个topic叫 ‘__consumer_offsets’，offset提交就是往这个topic发送一条消息，

消息格式是key value形式，key是由 groupId、主题名、分区号组成，消息体是位移值

及用户自定义数据和时间戳等。还有2种特殊的格式，一种是用于保存 Consumer Group 

信息的消息，用于注册group，另一种是 用于删除 Group 过期位移和删除 Group 的消息。

当kafka集群种第一台consumer启动时，便会创建__consumer_offsets主题，默认50个

分区和3个副本。

当提交方式是自动提交时，就算是当前consumer的offset已经不更新，kafka还是会自动

定期的往__consumer_offsets发送位移消息，所以得对位移主题的消息做定期删除，

假如对于同一个key有2条A和B，A早于B发送，那么A就是属于过期消息。

compact有点类似jvm gc的标记-整理，把过期消息删掉，把剩下的消息排列在一起
Kafka 提供了专门的后台线程定期地巡检待Compact 的主题，看看是否存在满足条件的

可删除数据，这个线程叫Log Cleaner，当我们发现位移主题日志过多的时候，可以

检查一下是否是这个线程挂了导致的

enable.auto.commit 默认即是true，

auto.commit.interval.ms 默认是5秒，表示kafka每5秒自动提交一次位移信息。

自动提交会有消息重复消费的问题，因为他是间隔时间提交一次，假如在间隔期间，

发生了Rebalance ，在Rebalance 之后所有的消费者必须从当前最新的offset开始

继续消费，那么上一次自动提交到Rebalance 的这段时间消费的数据的位移并没有

提交，所以会重复消费，即时我们通过减少 auto.commit.interval.ms 的值来提高提交频率，

那也仅仅是缩小了重复消费的时间窗口，所以我们看看能不能通过手动提交来避免重复消费。

commitSync()是consumer的同步api，手动提交的好处自然是我们可以控制提交的时机

和频率，由于是同步api，是会阻塞至broker返回结果才会结束这个阻塞状态，对于系统

而言，自然不想发生这种不是由于资源的限制导致的阻塞。

commitAsync()是consumer的异步api，commitAsync()不会阻塞，因此不会影响consumer的

tps，但是他的问题在于他无法重试，因为是异步提交，当因为网络或者系统资源阻塞

导致提交失败，那么他重试的时候，在这期间，consumer可能已经消费好多条消息

并且提交了，所以此时的重试提交的offset已然不是最新值了并没有意义，我们可以通过

异步和同步提交相结合，我们使用同步提交来规避因为网络问题或者broker端的gc导致的

这种瞬时的提交失败，进而通过重试机制而提交offset，使用异步提交来规避提交时的阻塞

前面的commitSync()和commitAsync()，都是consumer poll消息，把这些消息消费完，

再提交最新的offset，如果poll的消息很多呢？消费时间较长，假如中间系统宕机，岂不是

得从头再来一遍，所以kafka提供分段提交的api

commitSync(Map《TopicPartition, OffsetAndMetadata》) 

 commitAsync(Map《TopicPartition, OffsetAndMetadata》)

假设我们poll了一秒钟的数据，有5000条，我们可以通过计数器，累计到100条，

便通过分段提交api向kafka提交一次offset。