深度解析kafka broker网络组件初始化流程

àì夳堔傛蜴生んèń 2023-01-03 12:59 157阅读 0赞

> 原创不易，转载请注明出处  
> 阅读本文需要有java nio网络编程的基础

### 文章目录 ###

*   *   *  前言
         *  1.SocketServer初始化
         *  2.acceptor初始化
         *  3.processor初始化
         *  4.RequestChannel初始化
         *  5.RequestHandler初始化
         *  总结

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

### 前言 ###

在《[深度解析kafka broker网络模型运行原理][kafka broker]》一文中我们介绍了kafka broker 网络模型，并且大体上介绍了各个核心组件的运行原理，在第二小节中对比了一下RocketMQ网络模型与kafka的网络模型，发现其实差不多样子。本文主要是从源码角度出发，看看kafka那几个网络组件是怎样初始化的（acceptor，processor，RequestChannel，RequestHandler）

### 1.SocketServer初始化 ###

在源码中有个SocketServer类，你可以认为这SocketServer是kafka 网络服务器类，里面封装了kafka 网络各个组件，管理着它们的生命周期，大体上就是这样，我们来看下它的初始化源码  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70]  
其实它最上面的注释说的很明白了  
是个nio socket server，它的线程模型大体上是这个样子的：  
1个accept线程处理新连接  
accept线程下面有n个processor从socket里面读取请求啥的  
m个handler线程处理请求，返回响应给processor线程写回到socket中。

kafka broker是scala写的，然后创建SocketServer对象的时候就会将它那一堆成员初始化了。  
这里面有几个要注意的点：

1.  endpoints 这个就是你配置的那个listeners 参数 大体上就是这个样子`listeners=PLAINTEXT://:9092`
2.  numProcessorThreads 是processor线程数，默认是3个，这个参数配置`num.network.threads`
3.  maxQueuedRequests 这个参数与RequestChannel里面的请求队列有关，就是请求队列的大小，默认是500个 `queued.max.requests`管着配置
4.  totalProcessorThreads 这个就是总共processor线程数，这个与endpoints 的数量有关，默认endpoints我们就配置一个，所以默认总共就3个
5.  val requestChannel = new RequestChannel(totalProcessorThreads, maxQueuedRequests) 创建一个RequestChannel，我们下面再介绍
6.  processors 这个是processor线程数组，acceptors 是个map ， key是EndPoint ，value是acceptor

我们来看下这个SocketServer的startup 启动方法  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 1]  
我们重点看下红框里面的代码  
可以看到遍历这个endpoints（就是你监听的地址端口协议啥的，默认不管的就是一个9092）  
里面就是遍历创建processor对象，放到这个processors数组中，放到哪个位置是与endpoints循环有关的，如果你是第一次循环，就是放0，1，2 位置，如果是2次的话就是3，4，5位置。  
接着就是创建这个accept对象，可以看到是把创建的那几个processor对象传进去了，将这个acceptor对象放入map中，最后是创建kafka-socket-acceptor线程来运行这个acceptor对象，下面就是等这个acceptor 初始化完成，才会遍历下一个endpoint

其实从这里就可以得出这么一个结构来  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 2]  
你要是多个endpoint就是多个这个么个东西

### 2.acceptor初始化 ###

这个acceptor既然能放到一个线程中执行，它必然是个Runnable。  
我们来看下它的初始化  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 3]  
这个Acceptor 继承AbstractServerThread 这么一个抽象类，然后AbstractServerThread继承Runnable 接口  
![在这里插入图片描述][20210108142243588.png]  
看看初始化都干了些啥。

1.  nioSelector = NSelector.open() 这个就是打开了一个jdk nio的selector，这个NSelector就是jdk的Selector 类，这个scala整的虚头八脑的
2.  打开一个ServerSocketChannel，这个玩意也是jdk nio里面的  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 4]
3.  接着就是遍历自己分配的processor数组了，默认是分配了3个对象，分别创建线程来执行。

行了，初始化就3件事，1是打开selector，2是创建serverSocketChannel，3是默认3个processor线程来执行processor对象。

### 3.processor初始化 ###

我们来看下processor对象的初始化  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 5]  
这个processor也是个Runnable这个需要注意，它与acceptor一样都是继承AbstractServerThread  
这里也是3个比较重要的

1.  创建一个newConnections队列 ，这个对列是干啥的？acceptor新创建连接不是交给processor线程处理么，就是扔到这个processor对象的newConnections队列里面了，然后processor去这个队列中取出来，注册一下监听事件
2.  inflightResponses 这个map是 记录processor 还在发送中的响应，发送完了就移除去了
3.  创建一个kafka自己的 selector组件，这个KSelector里面封装了自己jdk nio的selector，然后里面一堆逻辑啥的。

### 4.RequestChannel初始化 ###

RequestChannel 是在SocketServer初始化的时候创建的，我们先看下参数

requestChannel = new RequestChannel(totalProcessorThreads, maxQueuedRequests)

这里这个totalProcessorThreads 就是总的processor线程数，如果你不配置的话其实就是3个，它与`listeners`参数配置的listener数量和`num.network.threads`参数配置有关  
maxQueuedRequests就是我们上面说的那个，默认是500个 `queued.max.requests`管着配置  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 6]  
这里面就看红框框出来的就可以，别的不用太多关注，都是些统计的东西。

1.  requestQueue 请求队列，默认大小是500个，processor从连接中读取的请求信息扔到这个requestQueue 队列中，然后一堆handler线程就从这个队列里面将请求取出来进行业务逻辑的处理。
2.  创建numProcessors 数量的responseQueue 数组，元素是个队列，每个processor 对应这个一个responseQueue 队列， handler将请求处理完成后，就会生成响应，就是将响应放到对应processor的responseQueue 队列的。不要问我handler是怎样知道应该塞哪个responseQueue 响应队列的，processor在接收到请求信息后，进行封装会把自己的一个编号放请求里面，然后在将请求放入请求队列中。

### 5.RequestHandler初始化 ###

我们可能会发现acceptor，processor，RequestChannel组件都在SocketServer中管理着，唯独RequestHandler没发现，其实严格意义来说这个RequestHandler它更偏向与业务处理的，但是网络模型少了它不行  
它是在KafkaServer 的startup方法初始化的，包括这个SocketServer 也是在

// apis
    apis = new KafkaApis(socketServer.requestChannel, replicaManager, groupCoordinator,
      kafkaController, zkUtils, config.brokerId, config, metadataCache, metrics, authorizer)
    
    // numIoThreads 默认是8个 ，这里面其实就是管理着几个handler（就是任务），默认是8个
    requestHandlerPool = new KafkaRequestHandlerPool(config.brokerId, socketServer.requestChannel, apis, config.numIoThreads)

可以看到我上面贴出来了一段代码，里面有KafkaApis 对象创建与KafkaRequestHandlerPool 对象的创建，为啥要把这个KafkaApis 对象创建拿出来呢？  
其实这里我想介绍下这个KafkaApis，它是用来分辨请求信息，看看这个请求是fetch请求还是produce请求还是metadata获取元数据请求，然后交给相对应的业务方法来处理。你可以理解为switch  
再来看下这个KafkaRequestHandlerPool ，它就是可以理解管着一堆RequestHandler线程的，毕竟是个pool。看下参数

1.  requestChannel 这个很熟悉，我们上面刚介绍的，RequestHandler 要从requestChannel的请求队列中拿请求信息，然后将响应信息放到对应的响应队列中。
2.  apis 这个上面刚介绍完，RequestHandler 拿到请求信息不知道找哪个方法处理，就找apis 来分辨是什么请求，找哪个方法处理。
3.  numIoThreads 这个是RequestHandler个数，也就是线程数。默认是8个 ， `num.io.threads`这个参数配置

看下KafkaRequestHandlerPool初始化代码  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 7]  
最重要的就是创建requestHandler对象与requestHandler线程，默认是8个。  
这个KafkaRequestHandlerPool 就是管着RequestHandler生命周期的，它里面就一个shutdown方法。

### 总结 ###

本文主要是介绍一下SocketServer ，acceptor，processor，RequestChannel，RequestHandler这些网络组件初始化，其中有几个比较重要的参数，包括processor线程数，requestHandler线程数，请求队列的大小。

[kafka broker]: https://inetyoung.blog.csdn.net/article/details/112307717
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70]: /images/20221119/80e9e6343583422d88d1185ae8ac2795.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 1]: /images/20221119/87cd25a379e44a498839129dafcd5ed0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 2]: /images/20221119/7d8a716f163446a8890b22c62c310940.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 3]: /images/20221119/185c957a840741099f983cc11000215e.png
[20210108142243588.png]: /images/20221119/23ae1e3d4ed94b71a8c0954b8b8129f3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 4]: /images/20221119/65f90cef9b0d4fb4a00533d808dc6ddd.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 5]: /images/20221119/f75528ff39cd4650abe7cc827021bde6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 6]: /images/20221119/a267c93615894d9582bc0ecf9efc78d9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 7]: /images/20221119/4b13c668fa7c4ecbb82bb7c720f84a14.png