Kafka服务端之日志存储

日志存储

基本概念

为了提高写入的性能，同一个分区中的消息是顺序写入的，这就避免了随机写入带来的性能问题。一个Topic可以划分成多个f分区，而每个分区又有多个副本。当一个分区的副本（无论是Leader 副本还是Follower 副本）被划分到某个Broker上时，Kafka就要在此Broker上为此分区建立相应的Log，而生产者发送的消息会存储在Log中，供消费者拉取后消费。

Kafka中存储的一般都是海量消息数据，为了避免日志文件太大，Log并不是直接对应于磁盘上的一个日志文件，而是对应磁盘上的一个目录，这个目录的命名规则是<topic_name>_<partition_id>，Log与分区之间的关系是一一对应的，对应分区中的全部消息都存储在此目录下的日志文件中。

Kafka通过分段的方式将Log分为多个LogSegment，LogSegment是一个逻辑上的概念，一个LogSegment对应磁盘上的一个日志文件和一个索引文件，其中日志文件用于记录消息，索引文件中保存了消息的索引。随着消息的不断写入，日志文件的大小到达一个阈值时，就创建新的日志文件和索引文件继续写入后续的消息和索引信息。日志文件的文件名的命名规则是[baseOffset].log，baseOffset是日志文件中第一条消息的offset。图4-13展示了一个Log的结构。

为了提高查询消息的效率，每个日志文件都对应一个索引文件，这个索引文件并没有为每条消息都建立索引项，而是使用稀疏索引方式为日志文件中的部分消息建立了索引。图4-14展示了索引文件与日志文件之间的对应关系。

介绍完Kafka日志存储的基本概念之后，下面来分析实现日志存储功能的相关代码。

FileMessageSet
在Kafka中使用FileMessageSet管理上文介绍的日志文件，它对应磁盘上的一个真正的日志文件。FileMessageSet继承了MessageSet抽象类，如图4-15（a）所示。MessageSet中保存的数据格式分为三部分，如图4-15（b）所示：8字节的offset值，4字节的size表示message data大小，这两部分组合成为LogOverhead，message data部分保存了消息的数据，逻辑上对应一个Message对象。

Kafka使用Message类表示消息，Message使用ByteBuffer保存数据，其格式及各个部分的含义如图4-16所示。

• ·CRC32：4个字节，消息的校验码。

• ·magic：1字节，魔数标识，与消息格式有关，取值为0或1。当magic为0时，消息的offset使用绝对offset且消息格式中没有timestamp部分；当magic为1时，消息的offset使用相对offset且消息格式中存在timestamp部分。所以，magic值不同，消息的长度是不同的。

• ·attributes：1字节，消息的属性。其中第0～2位的组合表示消息使用的压缩类型，0表示无压缩，1表示gzip压缩，2表示snappy压缩，3表示lz4压缩。第3位表示时间戳类型，0表示创建时间，1表示追加时间。

• ·timestamp：时间戳，其含义由attribute的第3位确定。

• ·key length：消息key的长度。

• ·key：消息的key。

• ·value length：消息的value长度。

• ·value：消息的value。

• MessageSet抽象类中定义了两个比较关键的方法：

这两个方法说明MessageSet具有顺序写入消息和顺序读取的特性。在后面对FileMessageSet和ByteBufferMessageSet的介绍过程中会介绍这两个方法的实现。

了解了MessageSet抽象类以及其中保存消息的格式，我们开始分析FileMessageSet实现类。FileMessageSet的核心字段如下所述。

• ·file：java.io.File类型，指向磁盘上对应的日志文件。
• ·channel：FileChannel类型，用于读写对应的日志文件。
• ·start和end：FileMessageSet对象除了表示一个完整的日志文件，还可以表示日志文件分片（Slice），start和end表示分片的起始位置和结束位置。
• ·isSlice：Boolean类型，表示当前FileMessageSet是否为日志文件的分片。
• ·_size：FileMessa_geSet大小，单位是字节。如果FileMessageSet是日志文件的分片，则表示分片大小（即end-start的值）；如果不是分片，则表示整个日志文件的大小。注意，因为可能有多个Handler线程并发向同一个分区写入消息，所有_size是AtomicInteger类型。

在FileMessageSet中有多个重载的构造方法，这里选择一个比较重要的方法进行介绍。此构造方法会创建一个非分片的FileMessageSet对象。在Window NTFS文件系统以及老版本的Linux文件系统上，进行文件的预分配会提高后续写操作的性能，为此FileMessageSet提供了preallocate的选项，决定是否开启预分配的功能。我们也可以通过FileMessageSet构造函数的mutable参数决定是否创建只读的FileMessageSet。

在FileMessageSet对象初始化的过程中，会移动FileChannel的position指针，这是为了实现每次写入的消息都在日志文件的尾部，从而避免重启服务后的写入操作覆盖之前的操作。对于新创建的且进行了预分配空间的日志文件，其end会初始化为0，所以也是从文件起始写入数据的。

介绍完FileMessageSet的构造过程，下面来分析其读写过程。FileMessageSet.append()方法实现了写日志文件的功能，需要注意的是其参数必须是ByteBufferMessageSet对象，ByteBufferMessageSet的内容后面介绍。下面是FileMessageSet.append()方法的代码：

查找指定消息的功能在FileMessageSet.searchFor()方法中实现。searchFor()的逻辑是：从指定的startingPosition开始逐条遍历FileMessageSet中的消息，并将每个消息的offset与targetOffset进行比较，直到offset大于等于targetOffset，最后返回查找到的offset。在整个遍历过程中不会将消息的key和value读取到内存，而是只读取LogOverhead（即offset和size），并通过size定位到下一条消息的开始位置。FileMessageSet.searchFor()方法的代码如下：

FileMessageSet.writeTo()方法是将FileMessageSet中的数据写入指定的其他Channel中，这里先了解此方法的功能，具体实现会在后面介绍“零复制”的时候一起介绍。FileMessageSet.read*()方法是从FileMessageSet中读取数据，可以将FileMessageSet中的数据读入到别的ByteBuffer中返回，也可以按照指定位置和长度形成分片的FileMessageSet对象返回。FileMessageSet.delete()方法是将整个日志文件删除。

FileMessageSet还有一个truncateTo()方法，主要负责将日志文件截断到targetSize大小。此方法在后面介绍分区中Leader副本切换时还会提到。下面是truncateTo()方法的具体实现：

FileMessageSet还实现了iterator()方法，返回一个迭代器。FileMessageSet迭代器读取消息的逻辑是：先读取消息的LogOverhead部分，然后按照size分配合适的ByteBuffer，再读取message data部分，最后将message data和offset封装成MessageOffset对象返回。此迭代器的实现与searchFor()方法类似。

ByteBufferMessageSet
MessageSet的另一个子类是ByteBufferMessageSet，FileMessageSet.append()方法的参数就是此类的对象。为什么必须append()方法的参数是ByteBufferMessageSet，而不是直接使用ByteBuffer呢?

在介绍MemoryRecords时提到，向MemoryRecords写入消息时，可以使用Compressor对消息批量进行压缩，然后将压缩后的消息发送给服务端。

在有些设计中，将每个请求的负载单独压缩后再进行传输，这种设计虽然可以减小传输的数据量，但是存在一个小问题，我们常见压缩算法是数据量越大压缩率越高，一般情况下每个请求的负载不会特别大，这就导致压缩率比较低。Kafka实现的压缩方式是将多个消息一起进行压缩，这样就可以保证较高的压缩率。而且在一般情况下，生产者发送的压缩数据在服务端也是以保持压缩状态进行存储的，消费者从服务端获取的也是压缩消息，消费者在处理消息之前才会解压消息，这也就实现了“端到端的压缩”。

压缩后的消息格式与非压缩的消息格式类似，但是分为两层，如图4-17所示。

压缩消息的key为null，所以图4-17没有画出key这部分；value中保存的是多条消息压缩数据。

创建压缩消息
在开始先回头看第2章介绍的Compressor的构造函数的代码：

通过MemoryRecords.append()方法不断写入消息并压缩的过程在第2章已经分析过了，这里不再赘述。当MemoryRecords写满，则会调用Compressor.close()方法，完成offset、size、CRC32等字段的写入，之后就可以发送到服务端了。Compressor.close()的实现如下：

细心的读者可能会问，服务端需要为每个消息分配offset，要对消息解压缩吗？这里的设计很巧妙，原理如下：

1）当生产者产生创建压缩信息的时候，对压缩消息设置的offset是内部offset（inner offset），即分配给每个消息的offset分别是0、1、2……请读者回顾第2章介绍的RecordBatch.tryAppend()方法。

2）在Kafka服务端为消息分配offset时，会根据外层消息中记录的内层压缩消息的个数为外层消息分配offset，为外层消息分配的offset是内层压缩消息中最后一个消息的offset值，如图4-18所示。

3）当消费者获取压缩消息后进行解压缩，就可以根据内部消息的、相对的offset和外层消息的offset计算出每个消息的offset值了。

ByteBufferMessageSet分析
介绍完生产者和消费者对压缩消息的处理过程，我们回到服务端，开始对ByteBufferMessageSet的分析，它底层使用ByteBuffer保存消息数据。ByteBufferMessageSet的角色和功能与MemoryRecords类似。ByteBufferMessageSet提供了三个方面的功能：

（1）将Message集合按照指定的压缩类型进行压缩，此功能主要用于构建ByteBufferMessageSet对象，通过ByteBufferMessageSet.create()方法完成。

提供迭代器，实现深层迭代和浅层迭代两种迭代方式。

3）提供了消息验证和offset分配的功能。
在ByteBufferMessageSet.create()方法中实现了消息的压缩以及offset分配，其步骤如下所示。
（1）如果传入的Message集合为空，则返回空ByteBuffer。
（2）如果要求不对消息进行压缩，则通过OffsetAssigner分配每个消息的offset，在将消息写入到ByteBuffer之后，返回ByteBuffer。OffsetAssigner的功能是存储一串offset值，并像迭代器那样逐个返回，OffsetAssigner的实现很简单，读者可参看源码学习。
（3）如果要求对消息进行压缩，则先将Message集合按照指定的压缩方式进行压缩并保存到缓冲区，同时也会完成offset的分配，然后按照压缩消息的格式写入外层消息，最后将整个外层消息所在的ByteBuffer返回。

我们回到本节开始的那个问题，FileMessageSet.append()方法会将ByteBufferMessageSet中的全部数据追加到日志文件中，对于压缩消息来说，多条压缩消息就以一个外层消息的状态存在于分区日志文件中了。当消费者获取消息时也会得到压缩的消息，从而实现“端到端压缩”。

OffsetIndex

为了提高查找消息的性能，从Kafka 0.8版本开始，为每个日志文件添加了对应的索引文件。OffsetIndex对象对应管理磁盘上的一个索引文件，与上一节分析的FileMessageSet共同构成一个LogSegment对象。

首先来介绍索引文件中索引项的格式：每个索引项为8字节，分为两部分，第一部分是相对offset，占4个字节；第二部分是物理地址，也就是其索引消息在日志文件中对应的position位置，占4个字节。这样就实现了offset与物理地址之间的映射。相对offset表示的是消息相对于baseOffset的偏移量。例如，分段后的一个日志文件的baseOffset是20，当然，它的文件名就是20.log，那么offset为23的Message在索引文件中的相对offset就是23-20 = 3。消息的offset是Long类型，4个字节可能无法直接存储消息的offset，所以使用相对offset，这样可以减小索引文件占用的空间。

Kafka使用稀疏索引的方式构造消息的索引，它不保证每个消息在索引文件中都有对应的索引项，这算是磁盘空间、内存空间、查找时间等多方面的折中。不断减小索引文件大小的目的是为了将索引文件映射到内存，在OffsetIndex中会使用MappedByteBuffer将索引文件映射到内存中。

介绍完了索引文件的相关概念后，我们来介绍OffsetIndex的字段。

• _file：指向磁盘上的索引文件。
• baseOffset：对应日志文件中第一个消息的offset
• ·mmap：用来操作索引文件的MappedByteBuffer。
• ·lock：ReentrantLock对象，在对mmap进行操作时，需要加锁保护。
• ·_entries：当前索引文件中的索引项个数。
• ·_maxEntries：当前索引文件中最多能够保存的索引项个数。
• ·_lastOffset：保存最后一个索引项的offset。

在OffsetIndex初始化的过程中会初始化上述字段，因为会有多个Handler线程并发写入索引文件，所以这些字段使用@volatile修饰，保证线程之间的可见性。初始化代码如下：

OffsetIndex提供了向索引文件中添加索引项的append()方法，将索引文件截断到某个位置的truncateTo()方法和truncateToEntries()方法，进行文件扩容的resize()方法。这些方法实际上都是通过mmap字段的相关操作完成的

OffsetIndex中最常用的还是查找相关的方法，使用的是二分查找，涉及的方法是indexSlotFor()和lookup()。值得注意的地方是，查找的目标是小于targetOffset的最大offset对应的物理地址（position）。下面是lookup()方法的代码：

OffsetIndexIndex的实现就介绍到这里。下一小节来分析LogSegment这个类。

LogSegment

为了防止Log文件过大，将Log切分成多个日志文件，每个日志文件对应一个LogSegment。在LogSegment中封装了一个FileMessageSet和一个OffsetIndex对象，提供日志文件和索引文件的读写功能以及其他辅助功能。

下面先来看LogSegment的核心字段。

• log：用于操作对应日志文件的FileMessageSet对象。
• ·index：用于操作对应索引文件的OffsetIndex对象。
• ·baseOffset：LogSegment中第一条消息的offset值。
• ·indexIntervalBytes：索引项之间间隔的最小字节数。
• ·bytesSinceLastIndexEntry：记录自从上次添加索引项之后，在日志文件中累计加入的Message集合的字节数，用于判断下次索引项添加的时机。
• ·created：标识LogSegment对象创建时间，当调用truncateTo()方法将整个日志文件清空时，会将此字段重置为当前时间

在LogSegment.append()方法中实现了追加消息的功能，可能有多个Handler线程并发写入同一个LogSegment，所以调用此方法时必须保证线程安全，在后面分析Log类时会看到相应的同步代码。另外，注意append()方法的参数，其第二个参数messages表示的是待追加的消息集合，第一个参数offset表示messages中的第一条消息的offset，如果是压缩消息，则是第一条内层消息的offset，如图4-20所示。

读取消息的功能由LogSegment.read()方法实现，它有四个参数。

• ·startOffset：指定读取的起始消息的offset。
• ·maxOffset：指定读取结束的offset，可以为空。
• ·maxSize：指定读取的最大字节数。
• ·maxPosition：指定读取的最大物理地址，可选参数，默认值是日志文件的大小。

在读取日志文件之前，需要将startOffset和maxOffset转化为对应的物理地址才能使用。这个转换在translateOffset()方法中实现，我们先通过一个例子来介绍其功能。现假设startOffset是1017，图4-21展示了将1017这个offset转换成对应的物理地址的过程。

（1）我们将absoluteOffset转换成Index File中使用的相对offset，得到17。通过OffsetIndex.lookup()方法查找Index File，得到(7,700)这个索引项，如图4-21中步骤①所示。

（2）根据(7,700)索引项，我们从MessageSet File中position=700处开始查找absoluteOffset为1017的消息，如图4-21中步骤②所示。

（3）通过FileMessageSet.searchFor()方法遍历查找FileMessageSet，得到(1018,800)这个位置信息，如图4-21中步骤③所示。

读者可能会问，我们的目标offset是1017，为什么会返回1018呢？在这个示例中，offset=1017的消息与其他的消息被压缩后一起构成了offset=1018这条外层消息，并存入了日志文件中。translateOffset()方法的代码如下：

了解了offset与物理地址之间的转换后，再来看read()方法就比较简单了。注意读取的结束位置由maxOffset、maxSize、maxPosition共同决定。LogSegment.read()方法的代码如下，其中省略了一些边界检查的代码：

LogSegment中还有一个值得注意的方法是recover()方法，其主要功能是根据日志文件重建索引文件，同时验证日志文件中消息的合法性。在重建索引文件过程中，如果遇到了压缩消息需要进行解压，主要原因是因为索引项中保存的相对offset是第一条消息的offset，而外层消息的offset是压缩消息集合中的最后一条消息的offset。

Log

Log是对多个LogSegment对象的顺序组合，形成一个逻辑的日志。为了实现快速定位LogSegment，Log使用跳表（SkipList）对LogSegment进行管理。

跳表是一种随机化的数据结构，它的查找效率和红黑树差不多，但是插入/删除操作却比红黑树简单很多。目前在Redis等开源软件中都能看到它的身影，在JDK中也提供了跳表的实现——ConcurrentSkipListMap，而且ConcurrentSkipListMap还是一个线程安全的实现，有兴趣的读者可以参考其源码。

在Log中，将每个LogSegment的baseOffset作为key，LogSegment对象作为value，放入segments这个跳表中管理，如图4-22所示。

例如，我们现在要查找offset大于6570的消息，可以首先通过segments快速定位到消息所在的LogSegment对象，定位过程如图4-22中的虚线所示。之后使用前面介绍的LogSegment.read()方法，先按照OffsetIndex进行索引，然后从日志文件中进行读取。

向Log中追加消息时是顺序写入的，那么只有最后一个LogSegment能够进行写入操作，在其之前的所有LogSegment都不能写入数据。最后一个LogSegment使用Log.activeSegment()方法获取，即segments集合中最后一个元素，为了描述方便，我们将此Segment对象称为“activeSegment”。随着数据的不断写入，当activeSegment的日志文件大小到达一定阈值时，就需要创建新的activeSegment，之后追加的消息将写入新的activeSegment。

介绍完了Log的基本原理后，来看一下Log类中的关键字段。

• ·dir：Log对应的磁盘目录，此目录下存放了每个LogSegment对应的日志文件和索引文件。
• ·lock：可能存在多个Handler线程并发向同一个Log追加消息，所以对Log的修改操作需要进行同步
• ·segments：用于管理LogSegment集合的跳表。
• ·config：Log相关的配置信息，具体配置项在具体代码中分析。
• ·recoveryPoint：指定恢复操作的起始offset，recoveryPoint之前的Message已经刷新到磁盘上持久存储，而其后的消息则不一定，出现宕机时可能会丢失。所以只需要恢复recoveryPoint之后的消息即可。
• ·nextOffsetMetadata：LogOffsetMetadata对象。主要用于产生分配给消息的offset，同时也是当前副本的LEO（LogEndOffset）。LEO的相关介绍请读者参考第1章。它的messageOffset字段记录了Log中最后一个offset值，segmentBaseOffset字段记录了activeSegment的baseOffset，relativePositionInSegment字段记录了activeSegment的大小。需要注意的是，为了保证在线程间的可见性，使用@volatile 修饰nextOffsetMetadata字段。

append()方法
向Log追加消息的功能是在Log.append()方法中实现的。Kafka服务端在处理生产者发来的ProducerRequest时，会将请求解析成ByteBufferMessageSet，并最终调用Log.append()方法完成追加消息，图4-23展示了这一调用关系。

Log.append()方法的大致流程如下：

（1）首先调用Log.analyzeAndValidateMessageSet()方法，对ByteBufferMessageSet中的Message数据进行验证，并返回LogAppendInfo对象。在LogAppendInfo中封装了ByteBufferMessageSet中第一个消息的offset、最后一个消息的offset、生产者采用的压缩方式、追加到Log的时间戳、服务端用的压缩方式、外层消息的个数、通过验证的总字节数等信息。
（2）调用Log.trimInvalidBytes()方法，清除未验证通过的Message。
（3）调用ByteBufferMessageSet.validateMessagesAndAssignOffsets()方法，进行内部压缩消息做进一步验证、消息格式转换、调整Magic值、修改时间戳等操作，并为Message分配offset。在ByteBufferMessageSet小节介绍过了，这里就不再赘述了。
（4）如果在validateMessagesAndAssignOffsets()方法中修改了ByteBufferMessageSet的长度，则重新验证Message的长度是否合法。
5）调用Log.maybeRoll()方法获取activeSegment，此过程可能分配新的activeSegment。
（6）将ByteBufferMessageSetSet中的消息追加到activeSegment中，通过调用LogSegment.append()方法的实现。
（7）更新当前副本的LEO，也就是Log.nextOffsetMetadata字段。
（8）执行flush()操作，将LEO之前的全部Message刷新到磁盘。

Log.append()方法的代码：

介绍了append()方法的骨架代码，下面具体分析其中每个方法的具体实现。步骤1中的Log.analyzeAndValidateMessageSet()方法主要功能是验证消息的长度、CRC32校验码、内部offset是否单调递增，这些验证都是对外层消息进行的，并不会解压内部的压缩消息。在append()方法的代码中我们也可以看到，如果需要进行offset分配，analyzeAndValidateMessageSet()方法返回的LogAppendInfo对象记录中的firstOffset、lastOffset甚至时间戳都会被修改。

Log.analyzeAndValidateMessageSet()方法的代码如下：

Log.maybeRoll()方法会检测是否满足创建新activeSegment的条件，如果满足则创建新activeSegment，然后返回当前的activeSegment。创建新activeSegment的条件有下面几个：

• 当前activeSegment的日志大小加上本次待追加的消息集合大小，超过配置的LogSegment的最大长度。
• ·当前activeSegment的寿命超过了配置的LogSegment最长存活时间
• 索引文件满了

maybeRoll()方法的代码如下，其中创建新activeSegment的逻辑在roll()方法中实现：

在Log.roll()方法最后使用KafkaScheduler线程池执行flush()方法，它是在JDK提供的ScheduledThreadPoolExecutor之上进行的封装和配置。在Kafka服务端有一部分定时任务是交由KafkaScheduler线程池执行的。KafkaScheduler线程的实现如下：

回到append()方法继续分析，步骤6调用了LogSegment.append()，请参考LogSegment小节的介绍。

最后来看flush()方法的原理，如图4-25所示，flush()方法会将recoverPoint~LEO之间的消息数据刷新到磁盘上，并修改recoverPoint值。

flush()方法的代码如下：

整个Log.append()方法的功能和实现到这里就分析完了。

read()方法

Log.read()方法实现了读取消息的功能，它实现的逻辑是：通过segments跳表，快速定位到读取的起始LogSegment并从中读取消息。注意，在Log.append()方法中通过加锁进行同步控制，在read()方法中并没有加锁操作，它在开始查询消息之前会将nextOffsetMetadata字段（@volatile修饰）保存成方法的局部变量，从而避免线程安全的问题。读者可以回顾updateLogEndOffset()方法的代码会发现每次更新updateLogEndOffset的时候，都是创建新的LogOffsetMetadata对象，而且LogOffsetMetadata中也没有提供任何修改属性的方法，可见LogOffsetMetadata对象是个不可变对象。

这里着重介绍（1）处的代码，为什么需要针对activeSegment的读取做特殊的处理呢？在Kafka的Bug列表中的[KAFKA-2477]描述了此Bug，下面介绍造成这个问题的主要原因。在Kafka 0.8版本中未修复这个问题，前面（1）处对应的代码如下：

简单描述这种场景：
（1）前面分析了append()方法的逻辑，在写线程调用append()方法时，会加锁写入，不会出现多个写线程的并发。现在按照append()方法的执行顺序将其分为两个操作：一是先分配offset并将Message追加到日志文件中，二是更新nextOffsetMetadata。

现在假设写线程在执行完第一步写入offset为18的消息后，CPU时间片到期，线程挂起，导致未对nextOffsetMetadata.messageOffset进行更新。

（2）其他的Follower 副本发来Fetch请求，读取offset为17以及之后的Message。按照Kafka 0.8版本的代码，Leader 副本会将offset为17、18的Message全部返回给Follower。

（3）Follower处理完offset为17、18两条Message，会继续请求offset为19的Message，此时，请求的startOffset>nextOffsetMetadata.messageOffset，Follower就会得到OffsetOutOfRangeException。Follower认为自己的Log出现了问题，会将此Log全部删除，并请求从Leader重新同步一份过来。

（4）之后，写线程重新执行，更新nextOfsetMetadata。

在海量数据的情况下，Kafka中的每个Log都很大的（在笔者的实践场景中，一个Log大约有15GB左右），如果多个Follower出现上述重新同步整个Log的情况，Leader副本所在的服服务器的I/O很快就会被占满，整个服务器都变得不可用。为了处理这个问题，就有了我们看到的对activeSegment的特殊处理，依然是上述场景，由于nextOffsetMetadata未更新，nextOffsetMetadata.relativePositionInSegment依然指向offset为17的Message的尾部，限制了Leader返回给Follower的消息。当Follower请求offset为18的消息时，返回的是消息集合是空。

LogManager
在一个Broker上的所有Log都是由LogManager进行管理的。LogManager提供了加载Log、创建Log集合、删除Log集合、查询Log集合等功能，并且启动了3个周期性的后台任务以及Cleaner线程（可能不止一个线程），分别是：log-flusher（日志刷写）任务、log-retention（日志保留）任务、recovery-point-checkpoint（检查点刷新）任务以及Cleaner线程（日志清理）。

下面介绍LogManager中各个字段的功能。

• ·logDirs：log目录集合，在server.properties配置文件中通过log.dirs项指定的多个目录。每个log目录下可以创建多个Log，每个Log都有自己对应的目录，不要混淆。LogManager在创建Log时会选择Log最少的log目录创建Log。
• ·ioThreads：为完成Log加载的相关操作，每个log目录下分配指定的线程执行加载。
• ·scheduler：KafkaScheduler对象，用于执行周期任务的线程池。与LogSegment小节介绍的执行flush()操作的scheduler是同一个对象。
• ·logs：Pool[TopicAndPartition, Log]类型，用于管理TopicAndPartition与Log之间的对应关系。使用的是Kafka自定义的Pool类型对象，底层使用JDK提供的线程安全的HashMap——ConcurrentHashMap实现
• ·dirLocks：FileLock集合。这些FileLock用来在文件系统层面为每个log目录加文件锁。在LogManager对象初始化时，就会将所有log目录加锁。
• ·recoveryPointCheckpoints：Map[File, OffsetCheckpoint]类型，用于管理每个log目录与其下的RecoveryPointCheckpoint文件之间的映射关系。在LogManager对象初始化时，会在每个log目录下创建一个对应的RecoveryPointCheckpoint文件。此Map的value是OffsetCheckpoint类型的对象，其中封装了对应log目录下的RecoveryPointCheckpoint文件，并提供对RecoveryPointCheckpoint文件的读写操作。RecoveryPointCheckpoint文件中则记录了该log目录下的所有Log的recoveryPoint。
• ·logCreationOrDeletionLock：创建或删除Log时需要加锁进行同步。

LogManager中的定时任务
在LogManager.startup()方法中，将三个周期性任务提交到scheduler中定时执行，并启动LogCleaner线程。LogManager.startup()方法的实现如下：

三个周期性任务的功能如表4-1所示。

log-retention任务通过周期性地调用LogManager.cleanupLogs()方法完成对符合条件的LogSegment的删除。cleanupLogs()方法的代码如下：

LogManager.cleanupExpiredSegments()方法会根据LogSegment的存活时长判断是否要删除LogSegment。

最后的Log.deleteSegment()方法完成了删除LogSegment的功能，其主要操作是清除segments跳表中的LogSegment对象，然后将日志文件和索引文件的后缀名改成“.deleted”，并创建一个删除这两个文件的任务，提交到scheduler线程池中异步执行。Log. deleteSegment()方法的代码如下：

介绍完LogManager.cleanupExpiredSegments()方法之后，再回来看LogManager.cleanupSegmentsToMaintainSize()方法，它会根据retention.bytes配置项的值与当前Log的大小判断是否删除LogSegment。

log-flusher任务会周期性地执行flush操作，其执行flush()方法的条件只有一个：Log未刷新时长是否大于此Log的flush.ms配置项指定的时长。

在每个log目录下都有唯一的一个RecoveryPointCheckpoint文件 ，其中记录此log目录下的每个Log的recoveryPoint值。RecoveryPointCheckpoint文件会在Broker启动时帮助Broker进行Log的恢复工作，具体恢复操作的流程后面会详细介绍。

recovery-point-checkpoint任务会周期性地调用LogManager.checkpointRecoveryPointOffsets()方法完成RecoveryPointCheckpoint文件的更新。checkpointRecoveryPointOffsets()方法的代码如下：

RecoveryPointCheckpoint文件的更新操作是在OffsetCheckpoint中实现的，其更新方式是：先将log目录下的所有Log的recoveryPoint写到tmp文件中，然后用tmp文件替换原来的RecoveryPointCheckpoint文件文件。OffsetCheckpoint.write()方法的是如下：

日志压缩
通过上面介绍的log-retention任务，Kafka服务端可以避免出现大量日志占满磁盘的情况。log-retention任务中配置的阈值非常灵活，可以对整个Broker设置全局配置值，也可以对某些特定的Topic配置特定值覆盖全局配置。

Kafka还提供了 “日志压缩”（Log Compaction）功能，通过此功能也可以有效地减小日志文件的大小，缓解磁盘紧张的情况。在很多实践场景中，消息的key与value的值之间的对应关系是不断变化的，就像数据库中的数据记录会不断被修改一样。如果消费者只关心key对应的最新value值，可以开启Kafka的日志压缩功能，服务端会在后台启动Cleaner线程池，定期将相同key的消息进行合并，只保留最新的value值。日志压缩的原理如图4-27所示，这里以key值为key3的消息为例进行说明，offset为3、6、10的三条消息按时间顺序依次被追加到Log中，在进行日志压缩时，只会保留key值为key3的最新消息（offset=10）。

我们已经知道，Log在写入消息时其实就是将消息追加到activeSegment的日志文件末尾。为了避免activeSegment成为热点，activeSegment不会参与日志压缩操作，而是只压缩其余的只读的LogSegment。在日志压缩过程中启动多条Cleaner线程，我们可以通过调整Cleaner线程池中的线程数量，优化并发压缩的性能，减少对整个服务端性能的影响。一般情况下，Log的数据量很大，为了避免Cleaner线程与其他业务线程长时间竞争CPU，并不会将除activeSegment之外的所有LogSegment在一次压缩操作中全部处理掉，而是将这些LogSegment分批进行压缩。

每个Log都可以通过cleaner checkpoint切分成clean和dirty两部分，clean部分表示的是之前已经被压缩过的部分，而dirty部分则表示未压缩的部分，如图4-28所示。现在假设Log中所有的消息都是非压缩消息，所有消息的offset都是连续的。日志压缩操作完成后，dirty部分消息的offset依然是连续递增的，而clean部分消息的offset是断断续续的。cleaner checkpoint与前面介绍的Log.recoveryPoint类似，保存在每个log目录对应一个的cleaner-offset-checkpoint文件中，由OffsetCheckpoint完成相应的读写操作。

每个Log需要进行日志压缩的迫切程度也不同，每个Cleaner线程只选取最迫切需要被压缩的Log进行处理。这里的“迫切程度”是通过cleanableRatio（dirty部分占整个Log的比例）决定的。

Cleaner线程在选定需要清理的Log后，首先为dirty部分的消息建立key与其last_offset（此key出现的最大offset）的映射关系，该映射通过SkimpyOffsetMap维护，后面会详细介绍SkimpyOffsetMap。然后重新复制LogSegment，只保留SkimpyOffsetMap中记录的消息，抛弃掉其他消息。经过日志压缩后，日志文件和索引文件会不断减小，Cleaner线程还会对相邻的LogSegment进行合并，避免出现过小的日志文件和索引文件。

最后值得注意的是，在日志压缩时，value为空的消息会被认为是删除此key对应的消息的标志，此标志消息会被保留一段时间，超时后会在下一次日志压缩操作中删除。

介绍完日志压缩的基本概念，来看一下日志压缩相关的实现类之间的依赖关系，如图4-29所示。

在LogCleaner中使用cleaners字段管理CleanerThread线程，通过startup()方法和shutdown()方法完成CleanerThread线程的启动和停止。

LogCleaner中的其他方法都直接委托给了LogCleanerManager对应的方法，代码不贴出来了。
LogCleanerManager主要负责每个Log的压缩状态管理以及cleaner checkpoint信息维护和更新。LogCleanerManager中各个字段的含义如下所述。

·checkpoints：Map[File, OffsetCheckpoint]类型，用来维护data数据目录与cleaneroffset-checkpoint文件之间的对应关系，与LogManager.recoveryPointCheckpoints 集合类似。
·inProgress：HashMap[TopicAndPartition, LogCleaningState]类型，用于记录正在进行清理的TopicAndPartition的压缩状态

...

到这里，Kafka的日志压缩功能以及其相关实现就介绍完了。下一节将回到LogManager中继续分析其初始化过程。

LogManager初始化
在LogManager的初始化过程中，除了初始化上述三个定时任务日志压缩的组件，还会完成相关的恢复操作和Log加载。首先调用LogManager.createAndValidateLogDirs()方法，保证每个log目录都存在并且可读，代码比较简单，就不贴出来了。之后会调用LogManager.loadLogs()方法加载log目录下的所有Log。这是LogManager初始化的重要过程，其步骤大致如下：

（1）为每个log目录分配一个有ioThreads条线程的线程池，用来执行恢复操作。
2）检测Broker上次关闭是否正常，并设置Broker的状态。在Broker正常关闭时，会创建一个“.kafka_cleanshutdown”的文件，这里就是通过此文件进行判断的。
（3）载入每个Log的recoveryPoint。
（4）为每个Log创建一个恢复任务，交给线程池处理。
（5）主线程阻塞等待所有的恢复任务完成。
（6）关闭所有在步骤1中创建的线程池。

LogManager.loadLogs()方法的代码如下：

从LogManager.loadLogs()方法的代码来看，只是创建了Log对象，并存入LogManager.logs集合进行管理。但是Log的初始化过程并不仅仅是创建一个对象而已，它会调用Log.loadSegments()方法.