Kafka是如何应用NIO实现网络通信的?（上）网络通信层SocketServer

网络通信层

Kafka网络通信层架构

kafka请求全处理流程

请求处理流程：

1. Clients或其他Broker通过Selector机制发起创建连接请求
2. Processor线程接收请求，并将其转换成可处理的Request对象
3. Processor线程将Request对象放入Request队列
4. KafkaRequestHandler线程从Request队列中取出待处理请求，并进行处理

1. KafkaRequestHandler线程将Response放回到对应Processor线程的Response队列
2. Processor线程发送Response给Request发送方

生产者-消费者模式的应用

Acceptor和Processor之间缓存SocketChannel newConnections队列，保证线程安全顺序。

Processor和KafkaRequestHandler之间缓存阻塞队列RequestChannel的Request全局队列和Response局部队列。

Kafka网络通信组成

• SocketServer

  核心，主要实现了Reactor模式，用于处理外部多个Clients（广义Clients，可能包含Producer、Consumer或其他Broker）的并发请求，并负责将处理结果封装进Response中，返还给Clients。

SocketServer是Kafka网络通信层中最重要的子模块。它的Acceptor线程、Processor线程和RequestChannel等对象，都是实施网络通信的重要组成部分。

KafkaRequestHandlerPool

I/O线程池，里面定义了若干个I/O线程，用于执行真实的请求处理逻辑。KafkaRequestHandlerPool线程池定义了多个KafkaRequestHandler线程，而KafkaRequestHandler线程是真正处理请求逻辑的地方。

两者共通处在于SocketServer中定义的RequestChannel对象和Processor线程。

在代码中，线程本质都是Runnable类型，不管是Acceptor类、Processor还是KafkaRequestHandler类。

相较于KafkaRequestHandler，Acceptor和Processor最多算请求和响应的“搬运工”。

AbstractServerThread类

这是Acceptor线程和Processor线程的抽象基类

Acceptor线程类

接收和创建外部TCP连接的线程。每个SocketServer实例只会创建一个Acceptor线程。

唯一作用创建连接，并将接收到的Request传递给下游的Processor线程。

Processor线程类

每个SocketServer实例默认创建若干个（num.network.threads）Processor线程。

负责

将接收到的Request添加到RequestChannel的Request队列

将Response返还给Request发送方

Processor伴生对象类

仅定义一些与Processor线程相关的常见监控指标和常量等，如Processor线程空闲率等。

ConnectionQuotas类

TooManyConnectionsException类

• SocketServer类

  实现了对以上所有组件的管理和操作，如创建和关闭Acceptor、Processor线程。

• SocketServer伴生对象类

  定义了一些有用的常量，同时明确了SocketServer组件中的哪些参数是允许动态修改的。

Acceptor线程

经典

React

or模式的Dispatcher接收外部请求并分发给下面的实际处理线程。在Kafka中，这个Dispatcher就是Acceptor线程。

参数

endPoint

定义的Kafka Broker连接信息，比如

PLAINTEXT://localhost:9092

sendBufferSize

recvBufferSize

如果在你的生产环境中，Clients与Broker的通信网络延迟很大（RTT>10ms），推荐增加控制缓冲区大小的两个参数：sendBufferSize和recvBufferSize，一般默认值100KB太小了。

Acceptor线程的自定义属性:

nioSelector

Java NIO库的Selector对象实例，也是后续所有网络通信组件实现Java NIO机制的基础

processors

网络Processor线程池。Acceptor线程在初始化时，需要创建对应的网络Processor线程池。Processor线程是在Acceptor线程中管理和维护的。

Processor相关API

• addProcessors

• 

• removeProcessors

• 

• 于是Acceptor类就具备Processor线程池管理功能。
• Acceptor类的run方法 - 处理Reactor模式中分发

• 

• Acceptor线程会先为每个入站请求确定要处理它的Processor线程

Acceptor线程使用Java NIO的Selector、SocketChannel循环轮询就绪的I/O事件（SelectionKey.OP_ACCEPT）。一旦接收到外部连接请求，Acceptor就指定一个Processor线程，并将该请求交由它，让它创建真正的网络连接。

Processor线程

• 源码

• 

• 执行流程

• 

• 每个Processor线程在创建时都会创建3个队列：可能是阻塞队列，也可能是一个Map对象