【云原生进阶之PaaS中间件】第三章Kafka-4.3.2-broker网络模型

作者：Gausst松鼠会 | 2024-02-12 20:45:00

踩

1 kafka网络模型运行原理

kafka broker 在启动的时候，会根据你配置的listeners 初始化它的网络组件，用来接收外界的请求，这个listeners你可能没配置过，它默认的配置是listeners=PLAINTEXT://:9092就是告诉kafka使用哪个协议，监听哪个端口，如果我们没有特殊的要求的话，使用它默认的配置就可以了，顶多是修改下端口这块。这个listeners是支持配置多套的，就是你可以监听多个端口，一个listener就对应着内部这么一套网络模型，我们就介绍一个listener的，多个其实都是一样的，就是对应着多套网络模型而已。

首先会创建一个accept 组件，这个组件对应着一个线程运行它，它主要是负责监听这个端口，打开一个selector，不用质疑就是jdk nio的东西，打开一个serverSocketChannel，然后专门监听accept事件，建立网络连接的。

接着会为这个accept组件建立创建几个processor组件，每个processor都对应这个一个线程运行，默认是3个，是由num.network.threads这个参数配置的，这几个processor专门是接收请求，发送响应的，每个processor都会打开一个selector用于事件监听。

当accept组件收到一个新连接请求的时候，会建立一个新连接，就会拿到一个socketChannel，将这个连接交给processor，processor拿到这个channel之后就会注册到对应的selector上面，监听它的read事件，然后后续关于这个连接发送的请求就由某个processor线程来处理，处理完之后再将响应写回到这个连接对应的channel中（事实上kafka中实际的业务逻辑处理不是processor处理的）。

到这里你以为就完事了？其实如果了解过netty的话，你会发现它跟netty的网络模型很像，netty也是一个线程组处理新连接的创建，一个线程组处理连接数据读取，处理，数据写回，其实这种网络模型就是reactor模型。你如果对netty有个比较深入的了解的话，你可能还会发现netty的pipeline中某个handler 是可以指定处理线程组的，好了我们接着介绍。

一开始的时候还会创建一个RequestChannel 组件，这个组件主要是几个队列，其中1个请求队列，专门存放请求的，还有几个是响应队列，响应队列的数量是与processor线程数量一样多，一个processor对应一个响应队列。

kafka broker 初始化的时候还会创建一个RequestHandlerPool组件，这个RequestHandlerPool从名字上也能猜出来是某个资源池，它初始化的时候，会创建几个RequestHandler线程放到这个Pool中，创建数量是由num.io.threads 参数配置的，默认是8个，它专门是干什么的呢？主要是从RequestChannel 的请求队列中获取请求，然后进行具体的业务处理，将处理完的响应再放到对应processor的响应队列中。

到这kafka broker 网络模型涉及到的核心组件以及主要作用介绍完了，下面我们将这些组件的运作流程串起来。

accept组件监听9092端口，当一个客户端发起建立连接请求时，accept会完成一个新连接的建立，拿到对应的channel，然后将这个channel交给某个processor线程，由它来监听处理这个连接的read ，write事件，就是请求与响应。当这个客户端发送一个请求的时候，processor线程就能监听到，拿到对应的请求信息后，将请求信息塞到RequestChannel 组件里面的请求队列中。

然后一堆RequestHandler 线程不停的从这个请求队列中获取请求信息，然后进行相应的业务逻辑处理，比如说发送消息的请求，它就会找到对应的parittion，写到对应partition在磁盘的文件中。处理完成业务逻辑后会有个处理结果需要告诉客户端，这个时候handler线程会将处理结果塞到对应processor的响应队列中。

processor会不停的从自己对应的那个响应队列中获取响应，然后写回给对应的客户端。

2 对比RocketMQ的网络模型

在RocketMQ中，使用的网络框架是netty，如果你了解netty更好，不了解也没关系，我这里将RocketMQ创建使用netty创建server的源码截出来，我们一起分析下。

我们上面也介绍过，netty也是reactor模型，一个boss线程组负责建立连接，一个work线程组负责连接请求信息读取与响应信息的写回，如果你是普通使用的话，work线程组还会处理你相应的业务逻辑。

当一个客户端请求建立连接的时候，boss线程组就会与客户端建立连接，然后将连接分配到work线程组某个线程上去，让work线程处理这个连接后续请求信息读取与响应信息的写回。到这里你会发现，boss线程组与kafka中的accept组件干的事差不多，然后work线程组与kafka中的processor线程干的事情差不多，可能唯一不一样的时候processor 将请求信息交出去了，然后这个work 线程自己处理了，其实在work线程处理的时候有个pipeline，你可以理解为有个链（理解为过滤器组都行，就是个责任链模式）要经过一堆自定义的handler，这些handler就可能有你的处理逻辑，比如说拆包粘包，编解码，业务逻辑处理，但是你可以为某个handler，或者是某些handler指定处理的线程池。也就是某个或者某些handler的业务逻辑交给这个线程池处理。要是将业务逻辑的handler单独交给某个线程池处理，是不是就对应了kafka的RequestHandler线程干的活，你就会发现他们的运行处理逻辑差不多了。接下来看看RocketMQ broker使用netty框架的网络模型是啥样的。

第一个红框是boss线程组，默认是1个线程，而且是写死的。 ------》对应着kafka accept 1个线程
boss线程组后面那个红框work线程组 eventLoopGroupSelector，默认是3个线程，可以通过serverSelectorThreads 参数配置。------》对应着 kafka processor线程 3个。
最下面那个框 defaultEventExecutorGroup 具体业务逻辑线程组，主要业务逻辑的处理是在NettyServerHandler中，默认是8个线程，可以通过serverWorkerThreads 参数配置。—》对应着kafka requestHandler线程 8个。

你会发现kafka与RocketMQ使用的网络模型相似，几乎一样，然后连默认的线程数都是一样的。其实不光网络模型相似，日志存储顺序写，充分利用os cache高速写。