深度解析kafka broker处理发送消息请求并写入磁盘

朱雀 2023-01-04 14:44 133阅读 0赞

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

### 文章目录 ###

*   *   *  前言
         *  1.requstHandler拿到请求找到对应处理方法
         *  2.ReplicaManager是怎样appendMessages的
         *  3.往partition的leader副本中追加消息
         *  4. Log是怎样append的
         *  5. segment的append方法

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

### 前言 ###

在《[深度解析kafka broker从连接建立到接收请求发送响应][kafka broker]》或者更往前介绍kafka broker 网络组件的文章中，我们介绍了broker处理与消息生产者（客户端）建立连接，processor接收请求，写回响应，requestHandler处理请求信息等流程，我们已经知道了一个请求过来是怎样运转的了，本文接着消息生产者发送消息请求这个场景继续介绍下requestHandler是怎样将消息写入到.log中，怎样写入稀疏索引到.index文件中的。

### 1.requstHandler拿到请求找到对应处理方法 ###

我们之前的文章其实介绍过requstHandler 是怎样从请求队列中拿出消息，然后调用apis来找到对应方法的，为了本文完整性，我们再回顾一下这里  
其实requestHandler是个Runnable ，我们直接看它的run方法就可以了。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70]  
其实就是2件事

1.  不断从请求队列中获取请求信息，直到获取到一个为止
2.  交给apis 的handle 方法处理这个请求信息，apis 就是KafkaApis  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 1]  
    我们这里是发送消息请求，所以走的是handleProducerRequest方法。  
    这个方法前半部分我们就不看了，就是验证权限这块的东西，再就是有个回调方法的定义，我们这里直接看重点  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 2]  
    这里是直接调用了副本管理器replicaManager的appendMessages方法。

### 2.ReplicaManager是怎样appendMessages的 ###

这里为啥要调用ReplicaManager 副本管理器呢？这个ReplicaManager 其实就是管理所有partition的副本的，不管是leader 副本还是follower副本，你往partition 中写消息其实就是往对应partition 的leader副本中写消息，然后其他的follower 副本再不停的从leader副本上获取消息。  
我们看看ReplicaManager 干了啥。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 3]  
首先是 验证你这个acks 合法性，只支持-1，0，1 这几个，如果你的acks不是这几个数的话，直接就响应错误（这块的每截图出来）  
介绍下-1，0，1 的含义

1.  0就是 客户端发送出去就完事了，不用等这个broker响应。
2.  1就是 leader 副本写入成功就可以了，不用管副本同步的事
3.  \-1 就是all的意思，要isr中`min.insync.replicas`个follower副本完成同步才可以

接着就是调用appendToLocalLog 方法追加到本地日志文件中。  
最后这块就是判断你这个acks，看看要不要等着其他follower副本同步完成，如果需要的话就创建延迟任务放到时间轮中，如果不需要的话就立马返回。  
我们这里看下appendToLocalLog 这个方法是是怎样将消息追加到本地日志文件中的  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 4]  
这个方法就是遍历这个messagesPerPartition map，因为你发送过来的消息是按照partition进行分组了，就是一个partition对应着一个消息集合。  
先是判断topic是不是kafka内置的topic，这里就是判断topic名字是不是“\_\_consumer\_offsets”，这个主要是kafka内部使用的，用来提交消费offset的。  
接着就是获取到对应的partition ，然后调用partition的appendMessagesToLeader 方法往leader副本中追加消息。

### 3.往partition的leader副本中追加消息 ###

我们来看下partition的appendMessagesToLeader 方法是怎样往leader 副本中追加消息的。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 5]  
首先是获取这个partition在本地的leader副本，如果获取不到，就抛出异常说本地没有这个partition的leader 副本，如果是能够获取到的话，从这个leader副本中拿到对应的Log对象，接着就是调用log的append 方法来往磁盘中写消息。  
最后就是有个maybeIncrementLeaderHW 方法来尝试增加下hw这个高水位，这个hw是与副本同步有关的。

### 4. Log是怎样append的 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 6]  
在这个log的append方法前面其实有很多代码，我们这里把它省略了，直接看比较重要的代码。  
我们可以看到这里是对代码块上了锁，也是多线程并发的时候会串行化，不会出现并发安全问题。  
先是获取一个offset ，这个offset其实就是这批消息起始offset ，接着就会为这批消息每条消息分配一个offset。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 7]  
这里就是获取一个日志段segment，在maybeRoll 这方法中如果是最后一个日志段log文件放不开这个堆消息或者index索引文件里面的索引数满了，就会roll，也就是创建一个新的日志段log文件与index索引文件，关于这块的代码我们在别的文章中介绍。  
接着就是调用日志段segment的append方法追加消息，最后是更新leo，判断要不要刷盘，是由`flush.messages`这个参数决定的。如果没刷盘的消息数量超过了这个阈值，就会进行刷盘操作，将内存中的日志内容刷到磁盘中。  
这里我们重点关注segment的append 方法

### 5. segment的append方法 ###

到这里其实就是将消息集追加写入到.log消息日志文件和写入稀疏索引。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 8]  
所谓的稀疏索引就是索引比较分散，并不是每条消息创建一个索引，而是隔着多少字节后创建一条索引，这里默认是4k，也就是写入超过4k消息后才会创建一个索引。  
先看下这个索引是怎样追加写的。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 9]  
可以看到，是先写入一个4个字节的offset ，需要注意的是它这个offset 是减去baseOffset之后，这个baseOffset就是这个日志段起始offset，是在整个partition中的一个偏移量。这里为啥要减去呢？其实是为了减少磁盘的占用，如果你要是存储partition offset的就要用个8字节的long类型来存储。  
接着追加写入一个4字节position，就是对应offset 所在的那个消息是在.log文件的哪个位置开始写的。  
最后就是变更这些计数的变量。

接着再来看看.log文件是怎样追加写消息集的  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 10]  
很简单，就是调用了messages的write方法，将消息写入到对应.log文件的channel中。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 11]  
其实就是循环追加写，到这里我们消息与对应稀疏索引写入磁盘就完事了（其实这里说写入磁盘是不严谨的，现在只是写入到了os cache中，需要你主动刷盘或者os 帮你刷盘才能将消息真正落实到日志文件中）

[kafka broker]: https://inetyoung.blog.csdn.net/article/details/112364682
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70]: /images/20221119/557667bd7adf4c54bce0deab60900ac4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20210110214154651.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20210110214358917.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20210110215026992.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/20210110221004811.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 5]: https://img-blog.csdnimg.cn/20210110225128710.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 6]: https://img-blog.csdnimg.cn/20210110225728814.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 7]: https://img-blog.csdnimg.cn/20210110230330563.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 8]: https://img-blog.csdnimg.cn/20210110231036451.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 9]: https://img-blog.csdnimg.cn/20210110231331804.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 10]: https://img-blog.csdnimg.cn/202101102318558.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 11]: https://img-blog.csdnimg.cn/20210110231959433.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv,size_16,color_FFFFFF,t_70