Kafka消息存储机制

尺有所短，寸有所长；不忘初心，方得始终。

Kafka 消息是以主题为单位进⾏归类，各个主题之间是彼此独⽴的，互不影响。每个主题⼜可以分为⼀个或多个分区。

• topic：消息的主题，可以理解为一个消息队列的名字
• partition：为了实现扩展性，一个topic分布到多个 broker（即服务器）上，一个topic可以分为多个partition，每个partition是一个有序的队列。
• segment：文件名一致的文件集合就称为 LogSement。partition物理上由多个segment组成
• message：每个segment文件中实际存储的每一条数据就是message
• offset：每个partition都由一系列有序的、不可变的消息组成，这些消息被连续的追加到partition中，partition中的每个消息都有一个连续的序列号叫做offset,用于partition唯一标识一条消息

一、Partition分区

• 每个分区在物理上对应一个文件夹，分区的命名规则为主题名后接“—”连接符，之后再接分区编号，分区编号从0开始，编号的最大值为分区总数减1

#分区目录文件
drwxr-x--- 2 root root 4096 Jul 26 19:35 kafka-topic-01-0
drwxr-x--- 2 root root 4096 Jul 24 20:15 kafka-topic-01-1
drwxr-x--- 2 root root 4096 Jul 24 20:15 kafka-topic-01-2

• 分区的每个副本在逻辑上可以抽象为一个日志对象，即分区副本与日志对象是相对应的
• 每个日志对象又可以划分为多个Segment文件，每个Segment文件包括一个日志数据文件和两个索引文件（偏移量索引文件和消息时间戳索引文件）

1.1、为什么要分区

通过分区实现水平扩展，提升性能

• 不分区：如果不分区每个topic的消息只存在一个broker上，所有的消费者都是从这个broker上消费消息，那么单节点的broker成为性能的瓶颈，存在单点问题。
• 分区：分区后消息分别存储在各个broker不同的partition上，消费者可以并行的从不同的broker不同的partition上读消息，实现了水平扩展。

1.2、分区文件下存了那些东西

每个分区下保存了很多文件，而概念上我们把他叫segment，即每个分区都是又多个segment构成的，其中文件名相同的index（索引文件），log（数据文件），timeindex（时间索引文件）统称为一个segment。

test-topic-0  
├── 00000000000000000001.index  
├── 00000000000000000001.log  
├── 00000000000000000001.timeindex  
├── 00000000000000001018.index  
├── 00000000000000001018.log  
├── 00000000000000001018.timeindex

1.3、为什么有了partition还需要segment

如果不引入segment，那么一个partition只对应一个文件（log），随着消息的不断发送这个文件不断增大，由于kafka的消息不会做更新操作都是顺序写入的，如果做消息清理的时候只能删除文件的前面部分删除，不符合kafka顺序写入的设计，如果多个segment的话那就比较方便了，直接删除整个文件即可保证了每个segment的顺序写入。

二、LogSement

• 分区⽇志⽂件中包含很多的 LogSegment
• Kafka ⽇志追加是顺序写⼊的
• LogSegment 可以减⼩⽇志⽂件的⼤⼩
• 进⾏⽇志删除的时候和数据查找的时候可以快速定位。
• ActiveLogSegment 是活跃的⽇志分段，拥有⽂件拥有写⼊权限，其余的 LogSegment 只有只读的权限。
• 每个 LogSegment 都有一个基准偏移量，用来表示当前 LogSegment 中第一条消息的 offset。

偏移量是⼀个 64 位的⻓整形数，固定是20位数字，⻓度未达到，⽤ 0 进⾏填补，索引⽂件和⽇志⽂件都由该作为⽂件名命名规则（00000000000000000000.index、00000000000000000000.timestamp、 00000000000000000000.log）。

如果⽇志⽂件名为 00000000000000000121.log ，则当前⽇志⽂件的⼀条数据偏移量就是 121（偏移量从 0 开始）。

2.1、日志数据文件

• Kafka将消息数据内容保存至日志数据文件中，该文件以该段的基准偏移量左补齐0命名，文件后缀为“.log”。
• 分区中的每条message由offset来表示它在这个分区中的偏移量，这个offset并不是该Message在分区中实际存储位置，而是逻辑上的一个值（8字节），
• offset唯一确定了分区中一条Message的逻辑位置，同一个分区下的消息偏移量按照顺序递增。

2.2、偏移量索引文件

• 偏移量索引文件和数据文件一样也同样也以该段的基准偏移量左补齐0命名，文件后缀为“.index”。

• 为了提高查找消息的效率，故而为分段后的每个日志数据文件均使用稀疏索引的方式建立索引。

  如果消息的消费者每次fetch都需要从1G大小（默认值）的日志数据文件中来查找对应偏移量的消息，效率低，在定位到分段后还需要顺序比对才能找到。

• 偏移量索引⽂件⽤于记录消息偏移量与物理地址之间的映射关系。

2.3、时间戳索引文件

• 时间戳索引文件是Kafka从0.10.1.1版本开始引入的的一个基于时间戳的索引文件，它们的命名方式与对应的日志数据文件和偏移量索引文件名基本一样，文件后缀为“.timeindex”。
• 每一条时间戳索引条目都对应了一个8字节长度的时间戳字段和一个4字节长度的偏移量字段，其中时间戳字段记录的是该LogSegment到目前为止的最大时间戳，后面对应的偏移量即为此时插入新消息的偏移量。
• 时间戳索引⽂件根据时间戳查找对应的偏移量。

2.4、补充说明

• 每个LogSegment中的日志数据文件大小均相等

  该日志数据文件的大小可以通过在Kafka Broker的config/server.properties配置文件的中的“log.segment.bytes”进行设置，默认为1G大小，在顺序写入消息时如果超出该设定的阈值，将会创建一组新的日志数据和索引文件

• Kafka索引⽂件是以稀疏索引的⽅式构造消息的索引，并不保证每个消息在索引⽂件中都有对应的索引项。

• 每当写⼊⼀定量的消息时，偏移量索引⽂件和时间戳索引⽂件分别增加⼀个偏移量索引项和时间戳索引项。

  通过修改 log.index.interval.bytes 的值，改变索引项的密度。

三、切分⽂件

日志文件和索引文件都会存在多个文件，组成多个 SegmentLog，当满⾜如下⼏个条件中的其中之⼀，就会触发⽂件的切分：

1. 当前⽇志分段⽂件的⼤⼩超过了 broker 端参数 log.segment.bytes 配置的值（1GB）。
2. 当前⽇志分段中消息的最⼤时间戳与当前系统的时间戳的差值⼤于 log.roll.ms 或 log.roll.hours 参数配置的值。log.roll.ms 的优先级 比log.roll.hours ⾼。默认只配置了 log.roll.hours 参数。
3. 偏移量索引⽂件或时间戳索引⽂件的⼤⼩达到 broker 端参数 log.index.size.max.bytes 配置的值。
4. 追加的消息的偏移量与当前⽇志分段的偏移量之间的差值⼤于 Integer.MAX_VALUE ，即要追加的消息的偏移量不能转变为相对偏移量。

• 为什么是Integer.MAX_VALUE

  Integer.MAX_VALUE  = 1024 * 1024 * 1024=1073741824 在偏移量索引⽂件中，每个索引项共占⽤ 8 个字节，并分为两部分，相对偏移量和物理地址。

  4 个字节刚好对应 Integer.MAX_VALUE ，如果⼤于 Integer.MAX_VALUE ，则无法⽤ 4 个字节表示。

• 相对偏移量：表示消息相对与基准偏移量的偏移量，占 4 个字节
• 物理地址：消息在⽇志分段⽂件中对应的物理位置，也占 4 个字节

四、索引

生产者生产消息之后消息内容保存在log⽇志⽂件中，消息封装为Record，追加到log⽇志⽂件末尾，采⽤的是顺序写模式，⼀个topic的不同分区，可认为是queue，顺序写⼊接收到的消息，通过索引查找数据。

• 偏移量索引：⽂件⽤于记录消息偏移量与物理地址之间的映射关系。
• 时间戳索引：⽂件则根据时间戳查找对应的偏移量。

4.1、偏移量索引

偏移量索引由相对偏移量和物理地址组成。

• 索引保存在index⽂件中，索引数据都是顺序记录 offset
• log⽇志默认每写⼊4K（log.index.interval.bytes），会写⼊⼀条索引信息到index⽂件中，因此索引⽂件是稀疏索引，它不会为每条⽇志都建⽴索引信息。
• log⽂件中的⽇志，是顺序写⼊的，由message+实际offset+position组成
• 索引⽂件的数据结构则是由相对offset（4byte）+position（4byte）组成，由于保存的是相对第⼀个消息的 相对offset，只需要4byte，可以节省空间，在实际查找后还需要计算回实际的offset。

稀疏索引：索引密度不⾼，但是offset有序，⼆分查找的时间复杂度为O(lgN)，如果从头遍历时间复杂度是O(N)。

如何查看偏移量为23的消息

Kafka 中存在⼀个 ConcurrentSkipListMap 来保存在每个⽇志分段，通过跳跃表⽅式，定位到在 00000000000000000000.index ，通过⼆分法在偏移量索引⽂件中找到不⼤于 23 的最⼤索引项，即 offset 20 那栏，然后从⽇志分段⽂件中的物理位置为320 开始顺序查找偏移量为 23 的消息。

4.2、时间戳索引

• 时间戳索引⽂件中每个追加的索引时间戳必须⼤于之前追加的索引项，否则不予追加

• 通过时间戳⽅式进⾏查找消息，需要通过查找时间戳索引和偏移量索引两个⽂件。

• 时间戳索引格式：前⼋个字节表示时间戳，后四个字节表示偏移量。

五、消息⽇志清理

Kafka 通过配置项log.cleanup.policy或者主题级别的配置项 cleanup.policy 来决定清理策略。

两种⽇志清理策略：

• ⽇志删除：配置项为delete （默认值），按照⼀定的删除策略，将不满⾜条件的数据进⾏数据删除

• ⽇志压缩：配置项为compact ，针对每个消息的 Key 进⾏整合，对于有相同 Key 的不同 Value 值，只保留最后⼀个版本。

  #启用删除策略
  log.cleanup.policy=delete 
  #直接删除，删除后的消息不可恢复。可配置以下两个策略：
  #清理超过指定时间清理：
  log.retention.hours=6
  #超过指定大小后，删除旧的消息：
  log.retention.bytes=1079441878
  # 检查日志段文件的间隔时间，单位是毫秒，以确定是否文件属性是否到达删除要求。
  log.retention.check.interval.ms=1000
  #文件在索引中清除后保留的时间一般不需要去修改【被标记.deleted的文件会在这个延迟时间后清除】
  log.delete.delay.ms =60000
  
  

5.1、⽇志删除策略

Kafka把topic中一个parition大文件分成多个小文件段，通过多个小文件段，定期清除或删除已经消费完文件，减少磁盘占用。

5.1.1 基于时间

• 删除时机⽇志删除任务会根据设定⽇志保留的时间节点，默认是 7 天。如果超过该设定值则进⾏删除。

  配置项为【log.retention.hours】【log.retention.minutes】【log.retention.ms】ms 优先级最⾼。

• Kafka 依据⽇志分段中最⼤的时间戳进⾏定位。

  ⾸先要查询该⽇志分段所对应的时间戳索引⽂件，查找时间戳索引⽂件中最后⼀条索引项，若最后⼀条索引项的时间戳字段值⼤于 0，则取该值，否则取最近修改时间。

  因为在写入消息时可以设置时间戳，所以⽇志⽂件的时间戳并不能真实的反应⽇志分段的最⼤时间信息。所以不能直接选最近修改时间。

• 删除过程

• 先从⽇志对象中所维护⽇志分段的跳跃表中移除待删除的⽇志分段。

  PS： 移除之前需保证没有线程对这些⽇志分段进⾏读取操作。

  如果活跃的⽇志分段中也存在需要删除的数据，Kafka 会先切分出⼀个新的⽇志分段作为活跃⽇志分段，该⽇志分段不删除，删除原来的⽇志分段。

• 在这些待删除的⽇志分段的所有⽂件添加上 .delete 后缀。

• 交由⼀个以 delete-file 命名的延迟任务来删除这些 .delete 为后缀的⽂件。延迟执⾏时间可以通过file.delete.delay.ms 进⾏设置

5.1.2 基于⽇志⼤⼩

• 删除时机

  日志删除任务会检查当前日志的大小是否超过设定的阈值retentionSize来寻找可删除的日志分段的文件集合deletableSegments

• 配置项为：log.retention.bytes 。
• 单个⽇志分段的⼤⼩配置项为：log.segment.bytes 。

• 删除过程

• 计算需要被删除的⽇志总⼤⼩ (当前⽇志⽂件⼤⼩（所有分段）减去log.retention.bytes值)。
• 从⽇志⽂件第⼀个 LogSegment 开始查找可删除的⽇志分段的⽂件集合。
• 延迟任务执⾏删除。

5.1.2 基于偏移量

• 删除时机

  基于日志起始偏移量的删除策略的判断依据是某日志分段的下一个日志分段的起始偏移量baseOffset是否小于等于logStartOffset，若是则可以删除此日志分段。

  假设logStartOffset等于60，日志分段1的起始偏移量为0，日志分段2的起始偏移量为11，日志分段3的起始偏移为23，那么我们通过如下动作收集可删除的日志分段的文件集合deletableSegments：

• 删除过程

• 从头开始遍历每个⽇志分段，⽇志分段1的下⼀个⽇志分段的起始偏移量为21，⼩于logStartOffset，将⽇志分段1加⼊到删除集合中。

• ⽇志分段 2 的下⼀个⽇志分段的起始偏移量为35,⼩于 logStartOffset，将⽇志分段2加⼊到删除集合中。

• ⽇志分段 3 的下⼀个⽇志分段的起始偏移量为57，⼩于logStartOffset，将⽇志分段3加⼊删除集合中。

• ⽇志分段4的下⼀个⽇志分段的其实偏移量为71，⼤于logStartOffset，则不进⾏删除。

5.2、⽇志压缩策略

⽇志压缩是Kafka的⼀种机制，可以提供较为细粒度的记录保留，对于具有相同的Key（不为null），⽽数据不同，只保留最后⼀条数据，

⽇志压缩特性在实时计算的异常容灾⽅⾯有很好的应⽤途径。⽐如在Spark、Flink中做实时计算时，需要⻓期在内存⾥⾯维护⼀些数据，这些数据可能是通过聚合了⼀天或者⼀周的⽇志得到的，这些数据⼀旦由于异常因素（内存、⽹络、磁盘等）崩溃了，从头开始计算需要很⻓的时间。⼀个⽐较有效可⾏的⽅式就是定时将内存⾥的数据备份到外部存储介质中，当崩溃出现时，再从外部存储介质中恢复并继续计算。

⽇志压缩⽅式的实现

⽇志压缩与key有关，确保每个消息的key不为null，首先主题的 cleanup.policy 需要设置为compact。

cleanup.policy = compact

Kafka的后台线程会定时将Topic遍历两次：

• 第一次记录每个key的hash值最后⼀次出现的偏移量
• 第⼆次检查每个offset对应的Key是否在后⾯的⽇志中出现过，如果出现了就删除对应的⽇志。⽇志压缩允许删除，除最后⼀个key之外，删除先前出现的所有该key对应的记录。在⼀段时间后从⽇志中清理，以释放空间。

压缩是在Kafka后台通过定时重新打开Segment来完成的，Segment的压缩细节如下图所示：

⽇志压缩可以过程：

• 任何滞留在日志head中的所有消费者能看到写入的所有消息；这些消息都是有序的offset。topic的使用min.compaction.lag.ms用来保障消息写入之前必须经过的最小时间长度，才能被压缩。

  min.compaction.lag.ms提供了消息保留在head（未压缩）的最少时间。

• 消息始终保持顺序，压缩永远不会重新排序消息，只是将旧数据删除，保留最新的。

• 消息的偏移量永远不会改变，它是⽇志中位置的永久标识符

• 由于删除日志与读取同时发生，消费者将优于删除。

默认情况下，启动⽇志清理器，若需要启动特定Topic的⽇志清理，配置⽇志清理器，

Kafka的⽇志压缩原理：就是定时把所有的⽇志读取两遍，写⼀遍，⽽CPU的速度超过磁盘完全不是问题，只要⽇志的量对应的读取两遍和写⼊⼀遍的时间在可接受的范围内，那么它的性能就是可以接受的。

六、磁盘存储

Kafka的高并发，高吞吐都有很明显的优势，原因在于：

• Kafka在设计的时候，采用了文件追加的方式来写入消息，即只能在日志文件的尾部追加新的消息，并且不允许修改已经写入的消息，这种方式属于典型的顺序写的操作，不需要寻址开销，所以就算是Kafka使用磁盘作为存储介质，对提升吞吐量有一定的作用。

• 除了顺序写之外，Kafka 的高并发写入主要是依靠页缓存和零拷贝两种技术实现的。

  页缓存技术主要用于消息写入 Kafka Broker 端的磁盘，零拷贝技术用于 Kafka Broker 将消息推送给下游消费者。

• kafka的两个过程：

• 网络数据持久化到磁盘 (Producer 到 Broker） 页缓存

• 磁盘文件通过网络发送（Broker 到 Consumer）零拷贝

6.1 零拷⻉

• kafka⾼性能，是多⽅⾯协同的结果，包括宏观架构、分布式partition存储、ISR数据同步、⾼效利⽤磁盘/操作系统特性。

• 零拷⻉并不是不需要拷⻉，⽽是减少不必要的拷⻉次数。通常是说在IO读写过程中。

6.1.1 传统 IO

传统 IO⽅式实现读取⽂件，socket发送过程：先读取、再发送，实际经过四次的拷⻉。

• 第⼀次：将磁盘⽂件，读取到操作系统内核缓冲区；

• 第⼆次：将内核缓冲区的数据，copy到application应⽤程序的buffer；

• 第三步：将application应⽤程序buffer中的数据，copy到socket⽹络发送缓冲区(属于操作系统内核的缓冲 区)；

• 第四次：将socket buffer的数据，copy到⽹络协议栈，由⽹卡进⾏⽹络传输。

实际IO读写，需要进⾏IO中断，需要CPU响应中断(内核态到⽤户态转换)，步骤 2, 3 是先等待 CPU 中断处理完毕后，再将数据读入内存。而每次的 IO 中断都会带来 CPU 的上下文切换。在现代操作系统中引入了直接内存访问 (DMA) 技术，它允许不同速度的硬件装置沟通，不需要依赖 CPU 的大量中断负载，数据的读写请求由 DMA 控制器接管，减少了 CPU 的负担。但四次copy中第二第三次是不必要的，数据可以直接从读缓冲区传输到套接字缓冲区。

以上四个步骤应用到Kafka中，在kafka没有使用零拷⻉，页缓存等优化的时候，Kafka Broker 将消息发送给下游的消费者流程：

• Kafka Broker 从磁盘中读取消息数据到系统内存；（内核模式）
• 系统内存拷贝数据到 Kafka Server 服务的缓存中（内核模式 -> 应用模式）；
• Kafka Server 服务将缓存中的消息数据复制到操作系统的 Socket 缓存中（应用模式 -> 内核模式）
• Socket 缓存将消息数据通过网卡发送出去（内核模式）

6.1.2 零拷贝(sendfile )

Kafka 在读消息时引入了零拷贝技术。零拷贝技术跳过了步骤 2 与步骤 3，跳过了两个上下文切换的步骤，将内核中的数据直接传输给 Socket 缓存，

• 磁盘数据通过DMA(Direct Memory Access，直接存储器访问)拷⻉到内核态 Buffer
• 直接通过 DMA 拷⻉到 NIC Buffer(socket buffer)，⽆需 CPU 拷⻉。

除了减少数据拷贝外，网络发送由一个 sendfile 调用完成，整个过程只有两次上下文切换，因此大大提高了性能。

上图中的 transferTo() 方法调用过程中的步骤如下：

• transferTo() 方法触发 DMA 引擎，将文件内容拷贝到一个读取缓冲区；（内核模式）
• 内核将数据拷贝到 Socket 缓冲区；（内核模式）
• DMA 引擎将数据从 Socket 缓冲传到网卡设备。（内核模式）

这样就将上下文切换次数减少到两次，即只留下必须的两次切换。但步骤 2 中的拷贝过程属于重复拷贝，可以将该步骤进行优化：

• transferTo() 方法触发 DMA 引擎，将文件内容拷贝到内核缓冲区；（内核模式）
• 关于数据的位置和长度的信息的描述符被追加到了 Socket 缓冲；
• 通过使用拷贝描述符的方法取代了拷贝数据，消除了剩下的最后一次 CPU 拷贝。
• DMA 引擎将数据从 Socket 缓冲传到网卡设备。（内核模式）

Kafka 的数据传输通过 TransportLayer 来完成，其⼦类 PlaintextTransportLayer 通过Java NIO 的 FileChannel 的 transferTo 和 transferFrom ⽅法实现零拷⻉。

public long transferFrom(FileChannel fileChannel, long position, long count) throws IOException {
        return fileChannel.transferTo(position, count, this.socketChannel);
}

6.2 页缓存

操作系统本身有一层缓存，叫做页缓存 (Page Cache)，又被称为 OS Cache，即为操作系统自己管理的缓存。页缓存可以将磁盘中的数据缓存到内存中，将对磁盘的访问转换为对内存的访问。以此⽤来减少对磁盘 I/O 的操作

Kafka 大量使用了页缓存，Kafka 在准备将消息写入磁盘中时，可以直接写入页缓存中，由操作系统自己刷盘。使得 Kafka 将消息写入磁盘的效率大幅提升。

Kafka接收来⾃socket buffer的⽹络数据，应⽤进程不需要中间处理，直接使⽤mmap内存⽂件映射进⾏持久化。

• mmap内存映射（Memory Mapped Files）
• 作⽤：将磁盘⽂件映射到内存, ⽤户通过修改内存就能修改磁盘⽂件。
• ⼯作原理：直接利⽤操作系统的Page来实现磁盘⽂件到物理内存的直接映射。完成映射之后你对物理内存的操作会被同步到硬盘上（操作系统在适当的时候）。
• 优点：通过 mmap，进程像读写硬盘一样读写内存（虚拟机内存）。提升I/O性能，省去了用户空间到内核空间复制的开销。
• 缺点：不可靠 ，写到 mmap 中的数据并没有被真正的写到硬盘，操作系统会在程序主动调用flush 的时候才把数据真正的写到硬盘。
• mmap的⽂件映射，在full gc时才会进⾏释放。当close时，需要⼿动清除内存映射⽂件，可以反射调⽤ sun.misc.Cleaner⽅法。

Kafka提供了⼀个参数 producer.type 来控制是不是主动flush

• 同步：Kafka写⼊到mmap之后就⽴即flush然后再返回Producer

  producer.type = sync
  
  

• 异步：写⼊mmap之后⽴即返回Producer不调⽤flush。

  producer.type = async
  
  

6.3 mmap 和 sendFile 的区别

• mmap 适合小数据量读写，sendFile 适合大文件传输。

• mmap 需要 4 次上下文切换，3 次数据拷贝；sendFile 需要 3 次上下文切换，最少 2 次数据拷贝。

• sendFile 可以利用 DMA 方式，减少 CPU 拷贝，mmap 则不能（必须从内核拷贝到 Socket 缓冲区）。

• 消费消息时：rocketMQ 使用了 mmap，kafka 使用了 sendFile。

七、 总结

• 顺序读写

• 磁盘顺序读写的速度400M/s，能够发挥磁盘最大的速度。

• 随机读写，磁盘速度慢的时候十几到几百K/s。

• 磁盘顺序读写在一定程度上比随机读写内存的速度还要快。

• kafka将来自Producer的数据，顺序追加在partition，partition就是一个文件，以此实现顺序写入。

• Consumer从broker读取数据时，因为自带了偏移量，接着上次读取的位置继续读，以此实现顺序读。

• 顺序读写，是kafka利用磁盘特性的一个重要体现。

• 零拷贝 sendfile(in,out)

• 数据直接在内核完成输入和输出，不需要拷贝到用户空间再写出去。

• kafka数据写入磁盘前，数据先写到进程的内存空间。

• mmap文件映射

• 虚拟映射只支持文件。

• 在进程的非堆内存开辟一块内存空间，和OS内核空间的一块内存进行映射。

• kafka数据写入、是写入这块内存空间，但实际这块内存和OS内核内存有映射，也就是相当于写在内核内存空间了，且这块内核空间、内核直接能够访问到，直接落入磁盘。

内核缓冲区的数据，flush就能完成落盘。