6.1.2 写入日志

柔情只为你懂 2023-01-12 15:52 84阅读 0赞

#### 6.1.2 写入日志 ####

> 服务端将生产者产生的消息集存储到日志文件，要考虑对消息集进行分段存储。如图6-3所示，服务端将消息追加到日志变件，并不是直接写入底层的文件，具体步骤如下。

(1)每个分区对应的日志对象管理了分区的所有日志分段。
    (2)将消息集迫加到当前活动的日志分段，任何时刻，都只会有一个活动的日志分段。
    (3)每个日志分段对应一个数据文件和索引文件，消息内容会追加到数据文件中。
    (4)操作底层数据的接口是文件通道，消息集提供一个WriteFullyTo()方法，参数是文件通道。
    (5)消息集(ByteBufferMessageSet)的WriteFullyTo()方法，调用文件通道的Write()方法，将底层包含消息内容的字节缓冲区(ByteBuffer)写到文件通道中。
    (6)字节缓冲区写到文件通道中，消息就持久化到日志分段对应的数据文件中了。

注意：在步骤(4)中，并不是直接调用文件通道的写方法，因为真正的消息内容还在消息集的ByteBuffer中。因此，妥让消息集提供一个写方法，接收的参数是文件通道；然后在它的写方法中，调用文件通道的写方法，并将字节缓冲区传入文件通道的写方法。  
![Å][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]

下面以消息集为入口，结合追加消息的流程，逐渐过被到日志、日志分段、数据文件、索引文件。

1.  消息集  
    在第2章中，生产者发送消息时，会在客户端将属于同一个分区的一批消息，作为一个生产请求发送给服务端。Java版本和IScala版本的生产者在客户端生成的消息集对象不一样，Java版本的消息内容本身就是字节缓冲区(ByteBuffer)，Scala版本则是消息集(Message仆。为了兼容两个版本，两者都要转换为底层是字节缓冲区的ByteBufferMessageSet对象。相关代码如下：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 1]

注意：Java版本生产者参与发送消息集的相关类有：RecordACCUmUlator(记录收集器)、RecordBatch(批记录kMemOryRecords(内存记录集)。最终发送出去的是内存记录集中的字节缓冲区，它和ScalaJt反本一样也有一个偏移量计数器。

如图6-4(左)所示，消息集中的每条消息(Message)都会被分配一个相对偏移盐，而每一批消息的相对偏移量都是从0开始的。图6-4(右)给出了一个示例，生产者写到分区Pl的第一批消息有4条消息，对应的偏移量是［0,1,2,3)；第二批消息有3条消息，对应的偏移量是［0,1,2)。客户端每次发送给服务端的一批消息，它的字节缓冲区只属于这一批消息，字节缓冲区不是共亭的数据结构。

如图6-5(上)所示，消息集中的每条消息由3部分组成：偏移盘、数据大小、消息内容。如图6-5(下)所示，每条消息除了保存消息的键值内容外，还保存一些其他数据，比如校验值、魔数、键的长度、值的长度等。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 2]

注意：Scala版本的消息格式在Message.scala类中，Java版本的消息格式在Record.java中。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 3]

消息集中每条消息的第一部分内容是偏移量。Kafka存储消息时，会为每条消息都指定一个唯一的偏移量。同一个分区的所有日志分段，它们的偏移量从0开始不断递增。不同分区的偏移量之间没有关系，所以说Kafka只保证r司一个分区的消息有序性，但是不保证跨分区消息的有序性。

消息集中每条消息的第二部分是当前这条消息的长度。消息长度通常不固定，而且在读取文件时，客户端可能期望直接定位到指定的偏移量。记录消息长度的好处是：如果不希望读取这条消息，只需要读取出消息民度这个字段的值然后跳过这些大小的字节，这样就可以定位到下一条数据的起始位置。

第三部分是消息的具体内容，和消息集的第二部分类似，每条消息的键值之前也都会先记录键的长度和值的长度。注意：消息、格式是在客户端定义的，消息集在传给服务端之前，就用ByteBufferMessageSet封装好。m~务端接收的每个分区消息、就是ByteBufferMessageSet。

如图6-6(左)所示，消息集的WriteMessage()方法将每条消息(Message填充到字节缓冲区中，缓冲区会暂存每个分区的一批消息。这个方法实际上是在客户端调用的，填充消息H才，会为这批消息设置从0开始递增的偏移量。如图6-6(右)所示，在服务端调用文件通道的写方法时，才会将消息集字节缓冲区的内容刷写到文件中。  
![在这里插入图片描述][20210414234857795.png]

客户端创建消息集中每条消息的偏移量，都还只是相对于本批次消息集的偏移岳。每一批消息的偏移量都是从0开始的，显然这个偏移量不能直接存储在日志文件中。对偏移量进行转换是在服务端进行的，客户端不需要做这个工作。为了获取到消息真正的偏移茧，必须知道存储在日志文件中“最近一条消息”的偏移量。如图6-7所示，假设日志文件中已经有5条消息了，最近一条消息的偏移盘是4［图6-7(左)］。再追加一个有5条消息的消息集时，这批消息的偏移盘从5开始［图6-7(中)］。再追加一个有4条消息的消息集时，这批消息的偏移量从10开始［图6-7(有汀，因为最近一条消息的偏移盐已经被更新为9。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 4]

注意：实际在计算并获取消息的偏移量时，采用“下一个偏移量”(nextOffset)，而不是“最近的偏移量”(lastOffset)。这样就可以直接使用“下一个偏移量”的佳，而不需妥在“最近偏移量”上再加一。

既然客户端消息的偏移量只是相对偏移量，都是从0开始的，为什么不在服务端计Jr消息偏移ill时直接设置？比如我们知道消息有5条，客户端不传每条消息的偏移盏，而是在服务端存储时，根据“最近的偏移量”，直接设置这5条消息的实际偏移量。实际上，客户端将消息填充到字节缓冲区时，消息的格式就已经确定下来了，只是偏移量的值还只是相对偏移量。

如图6-8(左)所示，服务端在存储消息时，可以直接修改字节缓冲区中每条消息的偏移量值，其  
他数据内容都不变，字节缓冲区的大小也并不会发生变化。而如图6-8(有)所示，客户端填充消息到字节缓冲区时没有写入相对偏移量。服务端存储消息时，由于最后要保存消息的偏移量，就需要在字节缓冲区每条消息的前面添加偏移量才行，这种方式会修改字节缓冲区的大小，原来的字节缓冲区就不能直接使用了。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 5]  
分析了消息集的构成后，下面来看消息集追加到日 志（ Log ） 的实现方式。

1.  日志追加  
    服务端将每个分区的消息追加到日志中，是以日志分段为单位的。当日志分段累加的消息达到阁值大小(文件大小达到lGB)时，会新创建一个日志分段保存新的消息，而分区的消息总是追加到最新的日志分段中。每个日志分段都有一个基准偏移量(segmentBaseOffset，或者口4baseOffset)，这个基准偏移址是分区级别的绝对偏移盘，而且这个值在日志分段中是固定的。有了这个基准偏移量，就可以计算出每条消息在分区中的绝对偏移量，最后把消息、以及对应的绝对偏移量写到日志文件中。

日志追加方法中的1’1essages参数是客户端创建的消息、集，这里面的偏移量是相对偏移量。在追加到日志分段时，validMessages变革已经是绝对偏移量了，具体步骤如下。

(1)对客户端传递的消息集进行验证，确保每条消息的(相对)偏移量都是单调递增的。
    (2)删除消息集中无效的消息。如果大小一致，直接返回1'1essages，否则会进行截断。
    (3)为有效消息集的每条消息分配(绝对)偏移量。
    (4)将更新了偏移聋值的消息集追加到当前日志分段中。
    (5)更新日志的偏移量(下一个偏移量)，必要时调用flush()方法刷写磁盘。

相关代码如下：  
![在这里插入图片描述][20210414235011743.png]![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 6]

上面代码最重要的是：volatile类型的nextOffsetMetadata变量。声明为volatile的变量被修改时，其他所有使用到此变量的线程都能立即见到变化后的值(称为“可见性”)。如图6-9所示，以nextOffsetMetadata为例，它的读写操作发生在服务端处理生产请求和拉取请求时，具体步骤如下。

(1)生产者发送消息集给服务端，服务端会将这一批消息追加到日志中μ
    (2)每条消息需要指定绝对偏移量，服务端会用nextOffsetMetadata的值作为起始偏移量。
    (3)服务端将每条带有偏移量的消息写入到日志分段中。
    (4)服务端会获取这一批消息中最后一条消息的偏移量，加上一后更新nextOffsetMetadata。
    (5)消费线程(消费者或备份副本)会根据这个变量的最新值拉取消息。一旦变量值发生变化，消费线程就能拉取到新写人的消息。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 7]

nextOffsetMetadata变it是一个关于臼志的偏移量元数据对象(LogOffset问etadata)。日志的偏移量元数据都是从当前活动的日志分段(act’lveSegment)获取相关的信息：系一条信息的偏移量、当前日志分段的基准偏移亟；、当前日志分段的大小。日志分段会在下一节分析，下面先来分析消息集在传到日志分段之前的处理流程。

1.  分析和验证消息集  
    对消息集进行分析和l验证，主要利用了Kafka中“分区的消息必须有序”这个特性。分析和验证方法的返回值是一个日志追加信息(LogAppendinfo)对象，该对象的内容包括：消息集第一条和最后一条消息的偏移盐、消息集的总字节大小、偏移量是否单调递增。

日志追加信息表示消息、集的概要信息，但并不包括消息内容。日志追加信息对象也是追加日志方法的最后返回值。服务端上层类(比如分区、副本管理器)调用追加日志的方法，期望得到这一批消息的概要信息，比如第一个偏移盐和最后一个偏移量。这样，它们就可以根据偏移量计算出－共追加了多少条消息(服务端接收的消息集和最后真正被追加的消息数量可能会不一样)。上层类甚至还可以做一些复杂的业务逻辑处理，比如根据最后一个偏移量判断被延迟的生产请求是否可以完成。相关代码如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 8]

前面说过，消息集对象中消息的偏移量是从0开始的相对偏移量，并且它的底层是一个字节缓冲区。那么要获得消息集中第一条消息和最后一条消息的偏移量，只能再把字节缓冲区解析出来，读取每一条消息的偏移盐。这里因为还要对每条消息进行分析和验证，所以读取消息是不可避免的。

注意：为了方便地获取字节缓冲区消息集中每条消息的内容及其对应的偏移量，消息集提供一个浅层选代器(shallowiterator)，迭代器的每个数据是一个MessageAndOffset对象，包括消息对象本身和偏移量。如果消息集有压缩，对应的迭代器是深层迭代器(deepiterator)。

分析消息集的每条消息时，都会更新最近的偏移盘(lastOffset)，但只会在分析第一条消息时更新起始偏移量(firstOffset)。判断消息集中所有消息的偏移量是否单调递增，只需要比较最近的偏移量和当前消息的偏移量。如果每次处理一条消息时，当前消息的偏移盘都比最近的偏移量值(上一条消息的偏移量)大，说明消息集是单调递增的。

对消息集的每条消息都验证和分析后，下一步要为消息分配绝对偏移茧，最后才能追加到日志分段。

1.  为消息集分配绝对偏移量  
    存储到日志文件中的消息必须是分区级别的绝对偏移量。为消息集分配绝对偏移盘时，以nextOffsetMetadata的偏移量作为起始偏移量。分配完成后还要更新nextOffsetMetadata的偏移毡’值。为了保证在分配过程中，获取偏移量的值并加一是一个原子操作，起始偏移量会作为原子变量传入validateMessagesAndAssignOffsets()方法。相关代码如下：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 9]

根据“1.消息集”中消息集的格式，为消息分配偏移量，实际上是更新每条消息的偏移量数据(offset)。消息的大小(size)和消息内容(Message)都不需要变动。现在的问题主要是：如何在字节缓冲区中定位到每条消息的偏移量所在位置。定位消息偏移量的方式有两种：一种是按照顺序完整地读取每条消息，这种方式代价比较大，我们实际上只需要更改偏移盘，不需要读取每条消息的实际内容；另一种是先读取Hi消息大小的值，然后计算下一条消息起始偏移量，最后直接用字节缓冲区提供的定位方法(pos’ition())直接定位到下一条消息的起始位置。

因为底层字节缓冲区和消息集对象是一一对应的，所以消息集中第一条消息的偏移盘一定是从字节缓冲区的位置。开始的。每条消息的长度计算方式是：8+4＋消息大小。其中，消息大小的值可以从第二部分读取。如表6-2所示，第一条消息中“消息的大小”存的值是3，表示消息本身的内容长度是3，整个消息占用的大小就是：8+4+3=15。  
![在这里插入图片描述][20210414235141531.png]

如图6-10所示，假设偏移量计数器初始值为I0(flPnextOffsetMetadata的值)，第一条消息的偏移量就等于10。分配第一条消息的偏移量时，修改前面8字节的内容为10。接下来要修改第二条消息的偏移量为ll，通过读取第一条消息的大小(等于3)，再加上12字节，就定位到第二条消息起始位置(等于15)。修改第三条消息的偏移量为12也是类似的，通过读取第二条消息的大小(等于5)，再加上12字节(第二条消息，总共占用了17字节)，就可以定位到第三条消息的起始位置(15再加上17，等于32)。以此类推，第四条消息的起始位置等于第三条消息占用的12字节再加上32，等于48。在写入每条消息的绝对偏移盘后，只会读取消息的大小，不会读取这条消息的实际内容。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 10]

消息集经过分配绝对偏移量后，才可以追加到日志分段中，日志分段接收消息集并写到文件中。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]: /images/20221022/d5e89d59c3014d3aa03096bfe1729e78.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 1]: /images/20221022/52fb16484be44ce790e47b02f81adcb1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 2]: /images/20221022/0427da2472e34327bc3aecf498955733.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 3]: /images/20221022/d0308b388a3c4d4784d53f3de0ac7c75.png
[20210414234857795.png]: /images/20221022/05eb1e5adc504bb3b72150081deb0ac1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 4]: /images/20221022/261ecd0b85df4056a85a557363958547.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 5]: /images/20221022/76995a65f45548d08011edbee46f3462.png
[20210414235011743.png]: https://img-blog.csdnimg.cn/20210414235011743.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 6]: /images/20221022/9783477b06184348b899706d6e300aab.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 7]: /images/20221022/9ab6cef466334fee9eebb8a1f6b3dac4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 8]: /images/20221022/af6bdbe4ff1342c299c510c80b1aab0d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 9]: /images/20221022/407fce5b4adf434db23f4c2f6c1ed56e.png
[20210414235141531.png]: /images/20221022/30c786f5132342f9a0f0ab16b35b6323.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 10]: /images/20221022/5f81b7726c1d40638e5afa30b2cf591a.png