深度解析kafka broker创建新日志段与更新LEO以及刷盘

心已赠人 2023-01-04 14:54 113阅读 0赞

> 原创不易，转载请注明出处

### 文章目录 ###

*   *   *  前言
         *  1.当前日志段放不下消息集是怎样创建新的日志段的
         *  2.写入完leader副本后是怎样更新leo的
         *  3.是怎样刷盘的
         *  总结

--------------------

### 前言 ###

在《[深度解析kafka broker处理发送消息请求并写入磁盘][kafka broker]》一文中我们介绍了包括requestHandler对发送消息请求处理以及消息集是怎样被追加写入到.log 文件，稀疏索引是怎样被写入到.index文件的，由于主流程的原因我们有些比较重要的细节没有深入的解析，本文就创建新的日志段，更新leo，刷盘的一些细节问题展开源码级剖析。

### 1.当前日志段放不下消息集是怎样创建新的日志段的 ###

回到Log类的append方法中，有`val segment = maybeRoll(validMessages.sizeInBytes)` 这么一行代码是获取日志段对象的，这里面如果当前的一个日志段的log文件或者index满了，没法继续写入新的消息集，就会将创建一个新的日志段。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70]  
先是获取最后一个日志段对象，判断如果当前日志段log文件大小在加上当前这个消息集大小大于了日志段最大大小（日志段最大大小是由`segment.bytes`参数控制的，然后默认是1个G）或者是这个日志段文件创建时间超过7天 或者是这个索引文件里面索引数大于了最大的那个，这个时候就要调用roll 方法来创建一个新的segment了  
否则的话就继续用最后那个就可以了。  
![。][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 1]  
可以看到其实就是创建对应的.log文件与.index索引文件，如果已经存在了就删除，然后就是创建这个日志段LogSegment 对象了，然后添加到log 对象的一个map缓存中，最后是更新leo。这个创建日志段LogSegment 对象 我们这里就不细讲了，其实就是拿到.log文件的channel 与.index文件的一个mmap channel。

### 2.写入完leader副本后是怎样更新leo的 ###

这里我们再回到Log类的append方法中，在将日志集追加到日志段文件中后，有这么一行代码`updateLogEndOffset(appendInfo.lastOffset + 1)`  
来更新这leo。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 2]  
其实就是将下一条消息的offset ， 这个日志段baseOffse，下一条消息在log文件的开始position封装到了一个LogOffsetMetadata 对象中了。  
这样子，下一批新消息过来的时候，就可以直接拿这个offset 当他们的起始offset。

### 3.是怎样刷盘的 ###

这里我们再回到Log类的append方法最后面的有这么一段代码

/// 如果没有刷盘的消息 大于这个 刷盘数量了 （这个刷盘数量默认是 个long 的最大值，就会进行刷盘）
    // 这个默认是不启用的
    if (unflushedMessages >= config.flushInterval)
     // 刷盘操作
     flush()

这个是用来刷盘的，我们在将消息集追加写入.log的channel与.index文件的channel中的时候，其实写入到了os cache中了，写入到操作系统一块内存缓存中了，我们可以设置这么一个刷盘策略，让它间隔多少条消息就将内存中的消息刷到磁盘上。  
你可以配置这个间隔阈值，默认是`flush.messages`参数控制，也是说没有刷盘的消息数大于了这个参数配置的阈值，就会进行刷盘动作，默认`flush.messages` 参数是无限大的，也就是不会管刷盘的事，完全由操作系统来控制。  
我们来看下和这个 flush() 方法是怎样刷盘的。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 3]  
其实就是遍历找到上次刷盘的那个日志段segment包括后面的所有日志段们，然后遍历调用segment.flush()方法进行刷盘。  
然后最后更新一下这次刷盘的offset，好留作下次刷盘使用，更新一下最后一次刷盘的时间。  
我们看下这里日志段对象是怎样刷盘的。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 4]  
这里其实调用了.log 文件包装对象FileMessageSet ，与.index文件包装对象OffsetIndex 的flush 方法。  
![在这里插入图片描述][2021011100061113.png]  
![在这里插入图片描述][20210111000623386.png]  
其实可以看到就是调用了对应文件channel的force 方法，这个样子就会将os cache中的消息强制刷到磁盘上去。

### 总结 ###

本文我们主要解析了在追加写消息的时候，如果当前日志段满了，是怎样roll到下个日志段的，其实就是创建一个新的。  
消息追加写入对应log 文件，index文件之后，LEO是怎样更新的，其实就是最后一条消息offset+1。  
刷盘的策略是怎么样的，然后是怎样进行刷盘的，其实底层就是调用了log文件，index文件对应channel 的force方法。

[kafka broker]: https://inetyoung.blog.csdn.net/article/details/112444268
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70]: /images/20221119/d0b0cad638314541b423ac9065868618.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 1]: /images/20221119/e1e3339b2a814d6c95febd8c0cfa4a2f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 2]: /images/20221119/cbf76463c5074cef9d95a289c15f16dd.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 3]: /images/20221119/9f032f88c87444b6a126a49462567fb4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 4]: /images/20221119/9a496aca76eb4f01a8156abfc51f9cc8.png
[2021011100061113.png]: /images/20221119/3fb02572f6154dfcaf7901439cc02ae0.png
[20210111000623386.png]: /images/20221119/1a4760bf0f8c410ab777eaeaff4e33dd.png