一、前言

Apache kafka 是一个开源的分布式事件流处理平台，由成千上万的公司用于高性能的数据管道流分析、数据集成和关键任务的应用程序。
从一开始，Kafka 就是为了解决大数据的实时日志流而生的, 每天要处理的日志量级在千亿规模。对于日志流的特点主要包括 1)、数据实时产生 2)、海量数据存储与处理，所以它必然要面临分布式系统遇到的高并发、高可用、高性能等挑战。Kafka对日志文件的存储和管理也是其基础，本文亦对其进行解析，希望对大家在使用kafka过程中有所帮助和启发。

二、日志存储结构

kafka 使用日志文件的方式来保存生产者和发送者的消息，每条消息都有一个 offset 值来表示它在分区中的偏移量。Kafka 中存储的一般都是海量的消息数据，为了避免日志文件过大，一个分片并不是直接对应在一个磁盘上的日志文件，而是对应磁盘上的一个目录，这个目录的命名规则是_。
比如创建一个名为firstTopic的topic，其中有3个partition，那么在 kafka 的数据目录（如/tmp/kafka-log）中就有 3 个目录，firstTopic_0~3
多个分区在集群中多个broker上的分配方法

• 将所有 N Broker 和待分配的 i 个 Partition 排序
• 将第 i 个 Partition 分配到第(i mod n)个 Broker 上

1.日志文件结构

2.topic

topic可以对应多个partition，而topic只是逻辑概念，不涉及到存储，partition才是物理概念，同一topic的不同partition可能分布在不同机器上

3.partition

partition是一个文件夹，其中包含多个segment，如果其中有n个segment，则共有2*n个文件，每个partition是一个有序的队列，partition中的每条消息都会分配一个有序的id，即offset1.3 segment 由一对文件组成，一个索引文件，一个数据文件
每个分片目录中，kafka 通过分段的方式将 数据 分为多个 LogSegment，一个LogSegment 对应磁盘上的一个日志文件（00000000000000000000.log）和一个索引文件(如上：00000000000000000000.index)，其中日志文件是用来记录消息的。索引文件是用来保存消息的索引。每个LogSegment 的大小可以在server.propertieslog.segment.bytes=1073741824 (设置分段大小,默认是1gb)选项进行设置。
文件命名规则，上一个segment在partition中的最大offset数值，即，比如00000000000000345678.log文件中的第一条消息的offset为345679 最大为64位的long，不足位数补0，如00000000000000345678.log 索引文件后缀.index，日志数据文件后缀.log
kafka 这种分片和分段策略，避免了数据量过大时，数据文件文件无限扩张带来的隐患，更有助于消息文件的维护以及被消费的消息的清理。

4.segment索引文件

• 存储一系列的元数据，每条元数据就是一条索引
• 元数据条数小于消息条数，是稀疏索引
• 元数据（即索引）构成

索引所指向的message在数据日志文件中的相对序号，即相对的offset，从1开始，该相对offset加上文件名当中的值就是该message在整个partition中的绝对offset了
索引所指向的message在数据日志文件中的位置，即文件游标，从0开始，方便直接指定游标打开数据文件

5.message结构

• offset 偏移量，消息的唯一标识，通过offset能找到唯一消息，类型long 8bytes
• MessageSize 消息长度，类型int32 4bytes
• crc32校验码， 4bytes，校验message
• magic， 表示本次发布kafka服务程序协议版本号 1byte
• attributes 独立版本，标识压缩类型，编码类型 1byte
• key length 4bytes 当key length=-1时，key字段可不写
• key 可选
• value byte payload 实际消息内容

为了提高查找消息的性能，kafka为每一个日志文件添加 了2 个索引文件：OffsetIndex 和 TimeIndex，分别对应.index以及.timeindex, *.TimeIndex 是映射时间戳和相对 offset的文件

message特点：

• message是无状态的，即不会标识是否已被消费过
• message不能同时被多个consumer来消费，可以等前一个消费完成，下一个继续消费。
• consumer可以同时消费多条message

6.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，最后查找到对应的消息以后返回。
注意：
可以利用offset在partition中查找，不能在整个topic中查找的，因为offset只保证在partition中唯一，有序

三、存储策略

1.顺序写

Kafka利用分段、追加日志的方式，在很大程度上将读写限制为顺序I/O（sequential I/O），这在大多数的存储介质上都很快。
人们普遍错误地认为硬盘很慢。然而，存储介质的性能，很大程度上依赖于数据被访问的模式。同样在一块普通的7200 RPM SATA硬盘上，随机I/O（random I/O）与顺序I/O相比，随机I/O的性能要比顺序I/O慢3到4个数量级。
此外，现代的操作系统提供了预先读和延迟写的技术，这些技术可以以块为单位，预先读取大量数据，并将较小的逻辑写操作合并成较大的物理写操作。

因此，顺序I/O和随机I/O之间的性能差异在闪存和其他固态非易失性介质中仍然很明显，不过它们在旋转存储，比如固态硬盘中的性能差异就没有那么明显。

2.页缓存

页缓存是操作系统用来作为磁盘的一种缓存，减少磁盘的I/O操作。
在写入磁盘的时候其实是写入页缓存中，使得对磁盘的写入变成对内存的写入。写入的页变成脏页，然后操作系统会在合适的时候将脏页写入磁盘中。在读取的时候如果页缓存命中则直接返回，如果页缓存 miss 则产生缺页中断，从磁盘加载数据至页缓存中，然后返回数据。并且在读的时候会预读，根据局部性原理当读取的时候会把相邻的磁盘块读入页缓存中。在写入的时候会后写，写入的也是页缓存，这样存着可以将一些小的写入操作合并成大的写入，然后再刷盘。而且根据磁盘的构造，顺序 I/O 的时候，磁头几乎不用换道，或者换道的时间很短。
缺点：
这样的写入存在数据丢失的风险，例如机器突然断电，那些还未刷盘的脏页就丢失了。不过可以调用 fsync 强制刷盘，但是这样对于性能的损耗较大。因此一般建议通过多副本机制来保证消息的可靠，而不是同步刷盘。

3.零拷贝

即使采用顺序写，但是频繁的 I/O 操作仍然会造成磁盘的性能瓶颈，所以 kafka还有一个性能策略：零拷贝。
消息从发送到落地保存，broker 维护的消息日志本身就是文件目录，每个文件都是二进制保存，生产者和消费者使用相同的格式来处理。

• 操作系统将数据从磁盘读入到内核空间的页缓存
• 应用程序将数据从内核空间读入到用户空间缓存中
• 应用程序将数据写回到内核空间到 socket 缓存中
• 操作系统将数据从 socket 缓冲区复制到网卡缓冲区，以便将数据经网络发出

这个过程涉及到 4 次上下文切换以及 4 次数据复制，并且有两次复制操作是由 CPU 完成。但是这个过程中，数据完全没有进行变化，仅仅是从磁盘复制到网卡缓冲区。
通过“零拷贝”技术，可以去掉这些没必要的数据复制操作，同时也会减少上下文切换次数。现代的 unix 操作系统提供一个优化的代码路径，用于将数据从页缓存传输到 socket；在 Linux 中，是通过 sendfile 系统调用来完成的。Java 提供了访问这个系统调用的方法：FileChannel.transferTo API，使用 sendfile，只需要一次拷贝就行，允许操作系统将数据直接从页缓存发送到网络上。所以在这个优化的路径中，只有最后一步将数据拷贝到网卡缓存中是需要的。

4.缓存机制

Kafka实际上是一个由磁盘支持的内存队列（受缓冲区/页面缓存大小的限制）。
Kafka在确认写操作之前并没有调用fsync，ACK的唯一要求是记录已经写入I/O缓冲区。但是，这种形式的写入是不安全的，因为副本的出错可能导致数据丢失，即使记录似乎已经被ACK。换句话说，与关系型数据库不同，仅写入缓冲区并不意味着持久性，保证Kafka持久性的是运行几个同步的副本，即使其中一个出错了，其他的（假设不止一个）将继续运行——假设出错的原因不会导致其他的副本也出错。因此，无fsync的非阻塞I/O方法和冗余的同步副本组合为Kafka提供了高吞吐、持久性和可用性。

四、日志格式演变

对于一个成熟的消息中间件来说，日志格式不仅影响功能的扩展，还关乎性能维度的优化。所以随着 Kafka 的迅猛发展，其日志格式也在不断升级改进中，Kafka 的日志格式总共经历了3个大版本：V0，V1和V2版本

1.V0 版本

V0 版本的消息格式主要存在于 Kafka 0.10.0.0 之前的版本，也是 Kafka 最早的消息版本，Kafka 的消息在 Kafka 的设计中被叫做 “Record”，我们也可以定位到 org.apache.kafka.common.record.Record 类，该类即是 Kafka 消息的类。
V0 版本消息格式示例如下：

• CRC 校验码，用于确保消息在传输过程中不会被篡改；
• magic：消息的版本号，这里 magic=0，表示 V0 版本；
• attributes：属性字段，V0 版本目前只使用该字段保存消息的压缩类型；
• key length：用于保存 key 字段长度，若 key 为空，则该字段为 -1；
• key：用于保存 key 值；
• value length：用于保存 value 字段长度，若 value 为空，则该字段为 -1；
• value：用于保存 value 值。

V0 版本的消息最小为 14 字节，小于 14 字节的消息会被 Kafka 视为非法消息

2.V1 版本

随着 Kafka 的不断迭代演进，用户发现 V0 版本的消息格式由于没有保存时间信息导致 Kafka 无法依据消息的具体时间作进一步判断，比如定期删除过期日志 Kafka 只能依靠日志文件的最近修改时间，这个时间很容易被外界干扰，比如在 linux 中执行了 touch 命令就会更改这个时间。
Kafka 0.10.0 版本，V1 版本的消息格式在 V0 版本的基础上增加了时间戳字段。
V1版本消息格式示例如下：

V1 版本的消息最小为 22 字节，小于 22 字节的消息会被 Kafka 视为非法消息。
总的来说比 V0 版本的消息大了 8 字节，如果还是按照 V0 那条消息计算，则在 V1 版本中它的总字节数为：22 + 8 = 30 字节。还需要注意的另一点差别就是 V1 版本中的 attribute 字段的第 4 位用于保存时间戳类型，当前时间戳类型有：

• CREATE_TIME：时间戳由 Producer 创建时间时指定；
• LOG_APPEND_TIME：时间戳由 broker 端写入消息时指定

3.V0/V1消息集合

V0、V1 版本的消息集合的设计没有任何区别，被称作“日志项”.格式如下：

message 字段也被 Kafka 称作浅层消息（shallow message），如果消息未进行压缩，那么该字段保存的消息即是它本身，如果消息进行压缩，Kafka 会将多条消息压缩在一起放入到该字段中。
值得注意的一点是：如果消息未被压缩，那么 offset 的值就是消息本身在分区日志中的 offset，如果多条消息被压缩放入到该字段中，则 offset 表示这批消息中最后一条消息在分区日志中的 offset。从这里我们也可以看出，在 V0、V1 版本的日志项中搜索位移是一件很困难的事情，我们需要解压并进行计算，代价非常高。
现在如果我们使用 V1 版本举例的那条消息放入消息集合中（未使用压缩），那么消息集合的大小为：8 + 4 + 30 = 42 字节。

4.V2 版本消息格式

针对 V0、V1 版本消息格式的缺陷，Kafka 在 0.11.0.0 版本对消息格式进行了大幅度重构，使用可变长度解决了空间使用率低的问题，增加了消息总长度字段，使用增量的形式保存时间戳和位移，并且把一些字段统一抽取到消息集合中。

Kafka 可变长度的具体做法借鉴了 Google ProtoBuffer 中的 Zig-zag 编码方式，这个我也没研究过，感兴趣的小伙伴可以研究下。但根据 Kafka 官方的描述，使用 Zig-zag 编码之后，例如一般的 key 只需要 1 字节保存即可，相比 V0、V1 版本需要 4 字节保存节省了 3 字节。
那么我们来总结一下 V2 版本具有哪些变化：

• 增加消息总长度字段：在消息格式的头部增加了消息总长度字段；
• 保存时间戳增量：不再使用 8 字节保存时间戳信息，更改成使用可变长度的字段保存增量信息，增量的起始时间戳值被保存在 V2 版本中的起始时间戳字段中，后面会降到；
• 保存位移增量：同理时间戳增量的做法；
• 增加消息头部：有了消息头部，就可以满足用户一些定制化需求了，用户可在消息头部保存一些自定义的元数据信息；
• 去除消息级别的 CRC 校验：移除消息级别的 CRC 校验，将 CRC 校验迁移到消息集合中。

还是以 V0 举例的消息为例，假设该条消息改成 V2 版本，那么该条消息的大小为：
1(sizeInBytes) + 1(attributes) + 1(timestamp) + 1(offset) + 1(key length) + 3(key) + 1(value length) + 5(value) + 1(headers length) = 15 字节。
可以看出，V2 版本的消息占用的空间会比 V0、V1 版本的消息要小很多

5.V2版本消息集合

V2 版本的消息批次，相比 V0、V1 版本主要有以下变动：

• CRC 值从消息中移除，被迁移到消息批次中；
• 增加了 PID、producer epoch、序列号等信息主要是为了支持幂等性以及事物引入的；
• 消息批次最小为 61 字节，相比 V0、V1 版本要大很多。

V2 版本主要是通过可变长度提高了消息格式的空间使用率，并将某些字段移到消息批次中，同时消息批次可容纳多条消息，从而在批量发送消息时，大幅度地节省了磁盘空间。

五、偏移量维护

offset的更新方式都有两种：自动提交和手动提交

1.自动提交（默认）

Kafka中偏移量的自动提交是由参数enable_auto_commit和auto_commit_interval_ms控制的，当enable_auto_commit=True时，Kafka在消费的过程中会以频率为auto_commit_interval_ms向Kafka自带的topic(__consumer_offsets)进行偏移量提交，具体提交到哪个Partation：Math.abs(groupID.hashCode()) % numPartitions。这种方式也被称为at most once，fetch到消息后就可以更新offset，无论是否消费成功。

2.手动提交

鉴于Kafka自动提交offset的不灵活性和不精确性(只能是按指定频率的提交)，Kafka提供了手动提交offset策略。手动提交能对偏移量更加灵活精准地控制，以保证消息不被重复消费以及消息不被丢失。
对于手动提交offset主要有3种方式：

• 同步提交
• 异步提交
• 异步+同步 组合的方式提交

同步提交：
提交失败的时候一直尝试提交，直到遇到无法重试的情况下才会结束，同步方式下消费者线程在拉取消息会被阻塞，在broker对提交的请求做出响应之前，会一直阻塞直到偏移量提交操作成功或者在提交过程中发生异常，限制了消息的吞吐量。
异步提交：
异步手动提交offset时，消费者线程不会阻塞，提交失败的时候也不会进行重试，并且可以配合回调函数在broker做出响应的时候记录错误信息。
对于异步提交，由于不会进行失败重试，当消费者异常关闭或者触发了再均衡前，如果偏移量还未提交就会造成偏移量丢失。
异步+同步：
针对异步提交偏移量丢失的问题，通过对消费者进行异步批次提交并且在关闭时同步提交的方式，这样即使上一次的异步提交失败，通过同步提交还能够进行补救，同步会一直重试，直到提交成功。通过finally在最后不管是否异常都会触发consumer.commit()来同步补救一次，确保偏移量不会丢失

六、日志清理策略

Kafka 将消息存储到磁盘中，随着写入数据不断增加，磁盘占用空间越来越大，为了控制占用空间就需要对消息做一定的清理操作。从上面 Kafka 日志结构分析中每一个分区副本（Replica）都对应一个 Log，而 Log 又可以分为多个日志分段（LogSegment），这样就便于 Kafka 对日志的清理操作。

Kafka提供了两种日志清理策略：
1.日志删除（Log Retention）：按照一定的保留策略直接删除不符合条件的日志分段（LogSegment）。
2. 日志压缩（Log Compaction）：针对每个消息的key进行整合，对于有相同key的不同value值，只保留最后一个版本。
这里我们可以通过 Kafka Broker 端参数 log.cleanup.policy 来设置日志清理策略，默认值为 “delete”，即采用日志删除的清理策略。如果要采用日志压缩的清理策略，就需要将 log.cleanup.policy 设置为 “compact”，这样还不够，必须还要将log.cleaner.enable（默认值为 true）设为 true。

如果想要同时支持两种清理策略， 可以直接将 log.cleanup.policy 参数设置为“delete，compact”

1.日志删除

在Kafka的日志管理器中会有一个专门的日志删除任务来周期性地检测和删除不符合保留条件的日志分段文件，这个周期可以通过broker端参数log.retention.check.interval.ms来配置，默认值为300000ms，即5分钟。
当前日志分段的保留策略有2种：

• 基于时间的保留策略
• 基于日志大小的保留策略。
• 基于日志起始偏移量

1.1 基于时间

日志删除任务会检查当前日志文件中是否有保留时间超过设定的阈值（retentionMs）来寻找可删除的日志分段文件集合（deletableSegments），retentionMs可以通过broker端参数log.retention.hours、log.retention.minutes和log.retention.ms来配置，其中 log.retention.ms 的优先级最高，log.retention.minutes 次之，log.retention.hours最低。
默认情况下只配置了log.retention.hours参数，其值为168，故默认情况下日志分段文件的保留时间为7天。
查找过期的日志分段文件，并不是简单地根据日志分段的最近修改时间lastModifiedTime来计算的，而是根据日志分段中最大的时间戳largestTimeStamp 来计算的。因为日志分段的lastModifiedTime可以被有意或无意地修改，比如执行了touch操作，或者分区副本进行了重新分配，lastModifiedTime并不能真实地反映出日志分段在磁盘的保留时间。
要获取日志分段中的最大时间戳 largestTimeStamp 的值，首先要查询该日志分段所对应的时间戳索引文件，查找时间戳索引文件中最后一条索引项，若最后一条索引项的时间戳字段值大于 0，则取其值，否则才设置为最近修改时间lastModifiedTime。

1.2 基于日志大小

日志删除任务会检查当前日志的大小是否超过设定的阈值（retentionSize）来寻找可删除的日志分段的文件集合（deletableSegments）。retentionSize可以通过broker端参数log.retention.bytes来配置，默认值为-1，表示无穷大。注意log.retention.bytes配置的是Log中所有日志文件的总大小，而不是单个日志分段（确切地说应该为.log日志文件）的大小。单个日志分段的大小由broker 端参数 log.segment.bytes 来限制，默认值为1073741824，即1GB。
基于日志大小的保留策略与基于时间的保留策略类似：
首先计算日志文件的总大小size和retentionSize的差值diff，即计算需要删除的日志总大小，然后从日志文件中的第一个日志分段开始进行查找可删除的日志分段的文件集合deletableSegments。

1.3 基于日志起始偏移量

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

1. 首先从头开始遍历每个日志段，日志段 1 的下一个日志分段的起始偏移量为20，小于logStartOffset的大小，将日志段1加入deletableSegments。
2. 日志段2的下一个日志偏移量的起始偏移量为35，也小于logStartOffset的大小，将日志分段2页加入deletableSegments。
3. 日志段3的下一个日志偏移量的起始偏移量为50，也小于logStartOffset的大小，将日志分段3页加入deletableSegments。
4. 日志段4的下一个日志偏移量通过对比后，在logStartOffset的右侧，那么从日志段4开始的所有日志段都不会加入deletableSegments。
5. 待收集完所有的可删除的日志集合后就可以直接删除了

2. 日志压缩策略

Kafka 还提供了“日志压缩（Log Compaction）”功能，通过这个功能可以有效的减少日志文件的大小，缓解磁盘紧张的情况，在很多实际场景中，消息的 key 和 value 的值之间的对应关系是不断变化的，就像数据库中的数据会不断被修改一样，消费者只关心 key 对应的最新的 value。因此，我们可以开启 kafka 的日志压缩功能，服务端会在后台启动Cleaner线程池，定期将相同的key进行合并，只保留最新的 value 值。日志的压缩原理如下图：

注意：
当启用压缩时，对批处理的影响特别明显，因为随着数据大小的增加，压缩通常会变得更有效。特别是在使用基于文本的格式时，比如JSON，压缩的效果会非常明显，压缩比通常在5x到7x之间。此外，记录的批处理主要作为一个客户端操作，负载在传递的过程中，不仅对网络带宽有积极影响，而且对服务端的磁盘I/O利用率也有积极影响。