kafka的文件储存机制,kafka消息结构

　　kafka的文件储存机制,kafka消息结构

　　00-1010消息存储路径数据碎片日志分段日志和索引文件内容分析。在分区中，通过偏移量找到消息处理日志的清理策略和压缩策略日志的清理策略。有两种日志压缩策略消息写入性能顺序写入零拷贝。

　　00-1010发送方将消息发送给代理后，消息如何持久化？

目录

kafka使用日志文件来保存生产者和发送者的消息，每个消息都有一个偏移值来表示它在分区中的偏移。

　　一般海量的消息数据都存储在Kafka中。为了避免日志文件太大，碎片不直接对应磁盘上的日志文件，而是对应磁盘上的目录。这个目录的命名规则是topic_name_partition_id。

　　例如，如果创建一个名为firstTopic的三分区主题，那么在kafka的数据目录(/tmp/kafka-log)中有三个目录，firstTopic-0~3。

　　多个分区在集群中多个broker上的分配方法

　　1.对要分配的所有nbrokers和I分区进行排序。

　　2.将第I个分区分配给第(i mod n)个代理

　　00-1010在每个碎片目录中，kafka通过分段将数据分成多个日志段，一个日志段对应磁盘上的一个日志文件(0000000000000.log)和一个索引文件(如上：00000000000000.index)，日志文件用于记录消息。索引文件是用于存储消息的索引。

　　可以在server.properties的log.segment.bytes=107370选项中设置每个日志段的大小(设置段大小，默认为1gb)。

　　段的索引文件和数据文件是一一对应的，成对出现。后缀。索引“和”。“日志”分别是段索引文件和数据文件。命名规则：partion整个世界中的第一个段从0开始，每个后续段文件的最后一个段文件中最后一条消息的偏移值递增。最大值为64位，长度为20位。没有数字是用0填充的。

　　第一个日志文件的最后一个偏移量是：5376，因此下一个段的文件被命名为0000000000000005376.log

　　对应的索引是000000000000005376.index

　　Kafka这种碎片分割策略，避免了数据量过大时数据文件无限膨胀带来的隐患，更有利于消息文件的维护和消耗消息的清理。

　　00-1010您可以通过以下命令查看kafka消息日志的内容

　　shkafka-run-class . shkafka . tools . dump logs-files/tmp/Kafka-logs/test-0/000000000000 . log-print-data-log输出结果为：

　　offset : 5376 position : 102124 create time : 1531477349287 is valid : true key size :-1 value size : 12 magic : 2 compress codec :N ONE producer :-1 producer epoch :-1 sequence 3:-1 is transactional : false header

　　您可以看到包含许多字段的消息，如下所示：

　　offset : 5371 position : 102124 create time : 1531477349286 is valid : true key size :-1 value size : 12 magic : 2 compress codec :N ONE producer :-1 producer epoch :-1 sequence 3:-1 is transactional 3: false header

　　>各字段的意义：

　　offset：记录号 ；position：偏移量；createTime：创建时间、keysize 和 valuesize 表示 key 和 value 的大小compresscodec：表示压缩编码payload：表示消息的具体内容为了提高查找消息的性能，kafka为每一个日志文件添加 了2 个索引文件：OffsetIndex 和 TimeIndex，分别对应*.index以及*.timeindex, *.TimeIndex 是映射时间戳和相对 offset的文件

　　查看索引内容命令：

 sh  kafka-run-class.shkafka.tools.DumpLogSegments  --files  /tmp/kafka-logs/test-0/00000000000000000000.index --print-data-log

索引文件和日志文件内容关系如下

　　如上图所示，index 文件中存储了索引以及物理偏移量。

　　log 文件存储了消息的内容。

　　索引文件中保存了部分offset和偏移量position的对应关系。

　　比如 index文件中 [4053,80899]，表示在 log 文件中，对应的是第 4053 条记录，物理偏移量（position）为 80899.

在 partition 中通过 offset 查找 message过程

根据 offset 的值，查找 segment 段中的 index 索引文件。由于索引文件命名是以上一个文件的最后一个offset 进行命名的，所以，使用二分查找算法能够根据offset 快速定位到指定的索引文件找到索引文件后，根据 offset 进行定位，找到索引文件中的匹配范围的偏移量position。（kafka 采用稀疏索引的方式来提高查找性能）得到 position 以后，再到对应的 log 文件中，从 position处开始查找 offset 对应的消息，将每条消息的 offset 与目标 offset 进行比较，直到找到消息比如说，我们要查找 offset=2490 这条消息，那么先找到00000000000000000000.index, 然后找到[2487,49111]这个索引，再到 log 文件中，根据 49111 这个 position 开始查找，比较每条消息的 offset 是否大于等于 2490。最后查找到对应的消息以后返回

日志的清除策略以及压缩策略

日志的清理策略有两个

根据消息的保留时间，当消息在 kafka 中保存的时间超过了指定的时间，就会触发清理过程根据 topic 存储的数据大小，当 topic 所占的日志文件大小大于一定的阀值，则可以开始删除最旧的消息。

通过 log.retention.bytes 和 log.retention.hours 这两个参数来设置，当其中任意一个达到要求，都会执行删除。默认的保留时间是：7 天

kafka会启动一个后台线程，定期检查是否存在可以删除的消息。

日志压缩策略

Kafka 还提供了日志压缩（Log Compaction）功能，通过这个功能可以有效的减少日志文件的大小，缓解磁盘紧张的情况，在很多实际场景中，消息的 key 和 value 的值之间的对应关系是不断变化的，就像数据库中的数据会不断被修改一样，消费者只关心 key 对应的最新的 value。

　　因此，我们可以开启 kafka 的日志压缩功能，服务端会在后台启动Cleaner线程池，定期将相同的key进行合并，只保留最新的 value 值。日志的压缩原理如下图：

消息写入的性能

顺序写

我们现在大部分企业仍然用的是机械结构的磁盘，如果把消息以随机的方式写入到磁盘，那么磁盘首先要做的就是寻址，也就是定位到数据所在的物理地址，在磁盘上就要找到对应的柱面、磁头以及对应的扇区；

　　这个过程相对内存来说会消耗大量时间，为了规避随机读写带来的时间消耗，kafka 采用顺序写的方式存储数据。

零拷贝

即使采用顺序写，但是频繁的 I/O 操作仍然会造成磁盘的性能瓶颈，所以 kafka还有一个性能策略：零拷贝

　　消息从发送到落地保存，broker 维护的消息日志本身就是文件目录，每个文件都是二进制保存，生产者和消费者使用相同的格式来处理。

　　在消费者获取消息时，服务器先从硬盘读取数据到内存，然后把内存中的数据原封不动的通过 socket 发送给消费者。

　　虽然这个操作描述起来很简单，但实际上经历了很多步骤。如下：

　　操作系统将数据从磁盘读入到内核空间的页缓存应用程序将数据从内核空间读入到用户空间缓存中应用程序将数据写回到内核空间到 socket 缓存中操作系统将数据从 socket 缓冲区复制到网卡缓冲区，以便将数据经网络发出这个过程涉及到 4 次上下文切换以及 4 次数据复制，并且有两次复制操作是由 CPU 完成。但是这个过程中，数据完全没有进行变化，仅仅是从磁盘复制到网卡缓冲区。通过零拷贝技术，可以去掉这些没必要的数据复制操作，同时也会减少上下文切换次数。

　　现代的 unix 操作系统提供一个优化的代码路径，用于将数据从页缓存传输到 socket；在 Linux 中，是通过 sendfile 系统调用来完成的。

　　Java 提供了访问这个系统调用的方法：FileChannel.transferTo API。

　　使用 sendfile，只需要一次拷贝就行，允许操作系统将数据直接从页缓存发送到网络上。

　　所以在这个优化的路径中，只有最后一步将数据拷贝到网卡缓存中是需要的

　　以上为个人经验，希望能给大家一个参考，也希望大家多多支持盛行IT。