首页
学习
活动
专区
工具
TVP
发布
精选内容/技术社群/优惠产品,尽在小程序
立即前往

kafka Log存储解析以及索引机制

本文设置到的配置项有

首先启动 kafka 集群,集群中有三台 Broker; 设置 3 个分区,3 个副本;

1 发送 topic 消息

启动之后kafka-client发送一个 topic 为消息szz-test-topic的消息

代码语言:javascript
复制
    public static void main(String[] args) {        Properties props = new Properties();        props.put("bootstrap.servers", "xxx1:9092,xxx2:9092,xxx3:9092");        props.put("acks", "all");        props.put("retries", 0);        props.put("batch.size", 16384);        props.put("linger.ms", 1);        props.put("buffer.memory", 33554432);        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");        Producer<String, String> producer = new KafkaProducer<>(props);        for(int i = 0; i < 5; i++){            producer.send(new ProducerRecord<String, String>("szz-test-topic", Integer.toString(i), Integer.toString(i)));        }        producer.close();    }

复制代码

发送了之后可以去log.dirs路径下看看

这里的 3 个文件夹分别代表的是 3 个分区; 那是因为我们配置了这个 topic 的分区数num.partitions=3; 和备份数offsets.topic.replication.factor=3; 这 3 个文件夹中的 3 个分区有LeaderFllower; 那么我们怎么知道谁是谁的 Leader 呢?

2 查看 topic 的分区和副本

代码语言:javascript
复制
bin/kafka-topics.sh  --describe --topic szz-test-topic --zookeeper localhost:2181

复制代码

可以看到查询出来显示分区 Partition-0在broker.id=0中,其余的是副本Replicas 2,1 分区 Partition-1在broker.id=1中,其余的是副本Replicas 0,2...

或者也可以通过 zk 来 查看leader在哪个 broker 上

代码语言:javascript
复制
 get /brokers/topics/src-test-topic/partitions/0/state

复制代码

代码语言:javascript
复制
[zk: localhost:2181(CONNECTED) 0] get /brokers/topics/szz-test-topic/partitions/0/state{"controller_epoch":5,"leader":0,"version":1,"leader_epoch":0,"isr":[0,1,2]}cZxid = 0x1001995bf

复制代码

3 分区文件都有啥

进入文件夹看到如下文件:

在这里插入图片描述

我们试试多发送一些消息,看它会不会生成新的 segment

代码语言:javascript
复制
public static void main(String[] args) {        Properties props = new Properties();        props.put("bootstrap.servers", "xxx1:9092,xxx2:9092,xxx3:9092");        props.put("acks", "all");        props.put("retries", 0);        props.put("batch.size", 163840);        props.put("linger.ms", 10);        props.put("buffer.memory", 33554432);        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        Producer<String, String> producer = new KafkaProducer<>(props);        for(int i = 0; i < 1200; i++){            //将一个消息设置大一点            byte[] log = new byte[904800];            String slog = new String(log);            producer.send(new ProducerRecord<String, String>("szz-test-topic",0, Integer.toString(i),  slog));        }        producer.close();    }

复制代码

在这里插入图片描述

从图中可以看到第一个 segment 文件00000000000000000000.log快要满log.segment.bytes的时候就开始创建了00000000000000005084.log了;并且.log.index.timeindex文件是一起出现的; 并且名称是以文件第一个 offset 命名的

  • .log 存储消息文件
  • .index 存储消息的索引
  • .timeIndex,时间索引文件,通过时间戳做索引

消息文件

上面的几个文件我们来使用 kafka 自带工具bin/kafka-run-class.sh 来读取一下都是些啥bin/kafka-run-class.sh kafka.tools.DumpLogSegments --files 00000000000000000000.log

最后一行:

代码语言:javascript
复制
baseoffset:5083  position: 1072592768  CreateTime: 1603703296169

复制代码

.index 消息索引

bin/kafka-run-class.sh kafka.tools.DumpLogSegments --files 00000000000000000000.index

最后一行:

代码语言:javascript
复制
offset:5083  position:1072592768

复制代码

.timeindex 时间索引文件

bin/kafka-run-class.sh kafka.tools.DumpLogSegments --files 00000000000000000000.timeindex

最后一行:

代码语言:javascript
复制
timestamp: 1603703296169 offset: 5083

复制代码

Kafka 如何查找指定 offset 的 Message 的

找了个博主的图 @lizhitao

比如:要查找绝对 offset 为 7 的 Message:

  1. 首先是用二分查找确定它是在哪个 LogSegment 中,自然是在第一个 Segment 中。
  2. 打开这个 Segment 的 index 文件,也是用二分查找找到 offset 小于或者等于指定 offset 的索引条目中最大的那个 offset。自然 offset 为 6 的那个索引是我们要找的,通过索引文件我们知道 offset 为 6 的 Message 在数据文件中的位置为 9807。
  3. 打开数据文件,从位置为 9807 的那个地方开始顺序扫描直到找到 offset 为 7 的那条 Message。

Kafka 中的索引文件,以稀疏索引(sparse index)的方式构造消息的索引,它并不保证每个消息在索引文件中都有对应的索引项。每当写入一定量(由 broker 端参数 log.index.interval.bytes 指定,默认值为 4096,即 4KB)的消息时,偏移量索引文件 和 时间戳索引文件 分别增加一个偏移量索引项和时间戳索引项,增大或减小 log.index.interval.bytes 的值,对应地可以缩小或增加索引项的密度。

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

leader-epoch-checkpoint

leader-epoch-checkpoint 中保存了每一任 leader 开始写入消息时的 offset; 会定时更新 follower 被选为 leader 时会根据这个确定哪些消息可用

4 参考文档

kafka官方文档

Kafka的Log存储解析

Kafka-工作流程,文件存储机制,索引机制,如何通过offset找到对应的消息

Broker配置文件详解

  • 发表于:
  • 本文为 InfoQ 中文站特供稿件
  • 首发地址https://www.infoq.cn/article/20d53420ea6622ead00cd8e13
  • 如有侵权,请联系 cloudcommunity@tencent.com 删除。

扫码

添加站长 进交流群

领取专属 10元无门槛券

私享最新 技术干货

扫码加入开发者社群
领券