消息中间件—简谈Kafka中的NIO网络通讯模型

摘要：不少人喜欢把RocketMQ与Kafka作对比，其实这两款消息队列的网络通讯层仍是比较类似的，本文就为你们简要地介绍下Kafka的NIO网络通讯模型，经过对Kafka源码的分析来简述其Reactor的多线程网络通讯模型和整体框架结构，同时简要介绍Kafka网络通讯层的设计与具体实现。

1、Kafka网络通讯模型的总体框架概述

Kafka的网络通讯模型是基于NIO的Reactor多线程模型来设计的。这里先引用Kafka源码中注释的一段话：

相信你们看了上面的这段引文注释后，大体能够了解到Kafka的网络通讯层模型，主要采用了 1（1个Acceptor线程）+N（N个Processor线程）+M（M个业务处理线程） 。下面的表格简要的列举了下（这里先简单的看下后面还会详细说明）：

线程数线程名线程具体说明1kafka-socket-acceptor_%xAcceptor线程，负责监听Client端发起的请求Nkafka-network-thread_%dProcessor线程，负责对Socket进行读写Mkafka-request-handler-_%dWorker线程，处理具体的业务逻辑并生成Response返回

Kafka网络通讯层的完整框架图以下图所示：

Kafka消息队列的通讯层模型—1+N+M模型.png

刚开始看到上面的这个框架图可能会有一些不太理解，并没关系，这里能够先对Kafka的网络通讯层框架结构有一个大体了解。本文后面会结合Kafka的部分重要源码来详细阐述上面的过程。这里能够简单总结一下其网络通讯模型中的几个重要概念：

（1）， Acceptor ：1个接收线程，负责监听新的链接请求，同时注册OP_ACCEPT 事件，将新的链接按照 "round robin" 方式交给对应的 Processor 线程处理；

（2）， Processor ：N个处理器线程，其中每一个 Processor 都有本身的 selector，它会向 Acceptor 分配的 SocketChannel 注册相应的 OP_READ 事件，N 的大小由 “num.networker.threads” 决定；

（3）， KafkaRequestHandler ：M个请求处理线程，包含在线程池—KafkaRequestHandlerPool内部，从RequestChannel的全局请求队列—requestQueue中获取请求数据并交给KafkaApis处理，M的大小由 “num.io.threads” 决定；

（4）， RequestChannel ：其为Kafka服务端的请求通道，该数据结构中包含了一个全局的请求队列 requestQueue和多个与Processor处理器相对应的响应队列responseQueue，提供给Processor与请求处理线程KafkaRequestHandler和KafkaApis交换数据的地方。

（5）， NetworkClient ：其底层是对 Java NIO 进行相应的封装，位于Kafka的网络接口层。Kafka消息生产者对象—KafkaProducer的send方法主要调用NetworkClient完成消息发送；

（6）， SocketServer ：其是一个NIO的服务，它同时启动一个Acceptor接收线程和多个Processor处理器线程。提供了一种典型的Reactor多线程模式，将接收客户端请求和处理请求相分离；

（7）， KafkaServer ：表明了一个Kafka Broker的实例；其startup方法为实例启动的入口；

（8）， KafkaApis ：Kafka的业务逻辑处理Api，负责处理不一样类型的请求；好比 “发送消息”、 “获取消息偏移量—offset” 和 “处理心跳请求” 等；

2、Kafka网络通讯层的设计与具体实现

这一节将结合Kafka网络通讯层的源码来分析其设计与实现，这里主要详细介绍网络通讯层的几个重要元素—SocketServer、Acceptor、Processor、RequestChannel和KafkaRequestHandler。本文分析的源码部分均基于Kafka的0.11.0版本。

一、SocketServer

SocketServer是接收客户端Socket请求链接、处理请求并返回处理结果的核心类，Acceptor及Processor的初始化、处理逻辑都是在这里实现的。在KafkaServer实例启动时会调用其startup的初始化方法，会初始化1个 Acceptor和N个Processor线程（每一个EndPoint都会初始化，通常来讲一个Server只会设置一个端口），其实现以下：

二、Acceptor

Acceptor是一个继承自抽象类AbstractServerThread的线程类。Acceptor的主要任务是监听而且接收客户端的请求，同时创建数据传输通道—SocketChannel，而后以轮询的方式交给一个后端的Processor线程处理（具体的方式是添加socketChannel至并发队列并唤醒Processor线程处理）。

在该线程类中主要能够关注如下两个重要的变量：

（1）， nioSelector ：经过NSelector.open()方法建立的变量，封装了JAVA NIO Selector的相关操做；

（2）， serverChannel ：用于监听端口的服务端Socket套接字对象；

下面来看下Acceptor主要的run方法的源码：

在上面源码中能够看到，Acceptor线程启动后，首先会向用于监听端口的服务端套接字对象—ServerSocketChannel上注册OP_ACCEPT 事件。而后以轮询的方式等待所关注的事件发生。若是该事件发生，则调用accept()方法对OP_ACCEPT事件进行处理。这里，Processor是经过 round robin 方法选择的，这样能够保证后面多个Processor线程的负载基本均匀。

Acceptor的accept()方法的做用主要以下：

（1）经过SelectionKey取得与之对应的serverSocketChannel实例，并调用它的accept()方法与客户端创建链接；

（2）调用connectionQuotas.inc()方法增长链接统计计数；并同时设置第（1）步中建立返回的socketChannel属性（如sendBufferSize、KeepAlive、TcpNoDelay、configureBlocking等）

（3）将socketChannel交给processor.accept()方法进行处理。这里主要是将socketChannel加入Processor处理器的并发队列newConnections队列中，而后唤醒Processor线程从队列中获取socketChannel并处理。其中，newConnections会被Acceptor线程和Processor线程并发访问操做，因此newConnections是ConcurrentLinkedQueue队列（一个基于连接节点的无界线程安全队列）

三、Processor

Processor同Acceptor同样，也是一个线程类，继承了抽象类AbstractServerThread。其主要是从客户端的请求中读取数据和将KafkaRequestHandler处理完响应结果返回给客户端。在该线程类中主要关注如下几个重要的变量：

（1）， newConnections ：在上面的 Acceptor 一节中已经提到过，它是一种ConcurrentLinkedQueue[SocketChannel]类型的队列，用于保存新链接交由Processor处理的socketChannel；

（2）， inflightResponses ：是一个Map[String, RequestChannel.Response]类型的集合，用于记录还没有发送的响应；

（3）， selector ：是一个类型为KSelector变量，用于管理网络链接；

下面先给出Processor处理器线程run方法执行的流程图：

Kafk_Processor线程的处理流程图.png

从上面的流程图中可以能够看出Processor处理器线程在其主流程中主要完成了这样子几步操做：

（1）， 处理newConnections队列中的socketChannel 。遍历取出队列中的每一个socketChannel并将其在selector上注册OP_READ事件；

（2）， 处理RequestChannel中与当前Processor对应响应队列中的Response 。在这一步中会根据responseAction的类型（NoOpAction/SendAction/CloseConnectionAction）进行判断，若为“NoOpAction”，表示该链接对应的请求无需响应；若为“SendAction”，表示该Response须要发送给客户端，则会经过“selector.send”注册OP_WRITE事件，而且将该Response从responseQueue响应队列中移至inflightResponses集合中；“CloseConnectionAction”，表示该链接是要关闭的；

（3）， 调用selector.poll()方法进行处理 。该方法底层即为调用nioSelector.select()方法进行处理。

（4）， 处理已接受完成的数据包队列—completedReceives 。在processCompletedReceives方法中调用“requestChannel.sendRequest”方法将请求Request添加至requestChannel的全局请求队列—requestQueue中，等待KafkaRequestHandler来处理。同时，调用“selector.mute”方法取消与该请求对应的链接通道上的OP_READ事件；

（5）， 处理已发送完的队列—completedSends 。当已经完成将response发送给客户端，则将其从inflightResponses移除，同时经过调用“selector.unmute”方法为对应的链接通道从新注册OP_READ事件；

（6）， 处理断开链接的队列 。将该response从inflightResponses集合中移除，同时将connectionQuotas统计计数减1；

四、RequestChannel

在Kafka的网络通讯层中，RequestChannel为Processor处理器线程与KafkaRequestHandler线程之间的数据交换提供了一个数据缓冲区，是通讯过程当中Request和Response缓存的地方。所以，其做用就是在通讯中起到了一个数据缓冲队列的做用。Processor线程将读取到的请求添加至RequestChannel的全局请求队列—requestQueue中；KafkaRequestHandler线程从请求队列中获取并处理，处理完之后将Response添加至RequestChannel的响应队列—responseQueue中，并经过responseListeners唤醒对应的Processor线程，最后Processor线程从响应队列中取出后发送至客户端。

五、KafkaRequestHandler

KafkaRequestHandler也是一种线程类，在KafkaServer实例启动时候会实例化一个线程池—KafkaRequestHandlerPool对象（包含了若干个KafkaRequestHandler线程），这些线程以守护线程的方式在后台运行。在KafkaRequestHandler的run方法中会循环地从RequestChannel中阻塞式读取request，读取后再交由KafkaApis来具体处理。

六、KafkaApis

KafkaApis是用于处理对通讯网络传输过来的业务消息请求的中心转发组件。该组件反映出Kafka Broker Server能够提供哪些服务。

3、总结

仔细阅读Kafka的NIO网络通讯层的源码过程当中仍是能够收获很多关于NIO网络通讯模块的关键技术。Apache的任何一款开源中间件都有其设计独到之处，值得借鉴和学习。对于任何一位使用Kafka这款分布式消息队列的同窗来讲，若是可以在必定实践的基础上，再经过阅读其源码能起到更为深刻理解的效果，对于大规模Kafka集群的性能调优和问题定位都大有裨益。

对于刚接触Kafka的同窗来讲，想要本身掌握其NIO网络通讯层模型的关键设计，还须要不断地使用本地环境进行debug调试和阅读源码反复思考。

文章来源：https://segmentfault.com/a/1190000016555478

推荐阅读：https://www.roncoo.com/course/view/a398edee6308413f904f9c82b1dfc122