写在前面

这篇文章浅谈 Broker 的网络架构，希望能给他做一个直观的展示。

Reactor 模式

Kafka Broker 的网络处理采用的是 Reactor 模式，所以需要先了解一下 Reactor 模式。而理解 Reactor 架构，网络上已经有了很多详细的中文说明资料。我在这里只做一些简单的描述，便于保持一个完整的文章结构。

要解决的问题

任何软件架构都是要解决某一个问题的，那么 Reactor 也不例外。它要解决的问题就是：如何处理高并发情况下的客户端与服务端连接请求问题。这个问题细拆有以下要点：

1. 角色：常见的 C/S 架构，即客户端与服务端两个角色。

2. 操作：客户端向服务端发起的连接与请求。在网络通信中，连接与请求是分开的。具体如下：

3. 连接是为了保证请求和响应能够稳定可靠的传输；请求是为了从服务端获取或操作资源

4. 连接可以是短暂或持久的。请求也可以是单一或多个的

5. 建立了连接并不一定有请求（可能客户端还未准备好请求的数据）；但客户端发送请求肯定需要提前建立好连接。

现在需要把视角聚焦于服务端，需要从服务端思考，在多个客户端高并发请求的情况下，我们如何解决上面的问题。

粗略的演进

最直观的思考

从服务端的角度做最直观的思考就是：针对客户端的每一次连接与请求，都启动专门的线程/进程（后面方便以线程为主）去处理连接、请求、响应、断连操作。具体流程如下：

1. 客户端建立连接。服务端启动线程去建立与客户端的连接

2. 客户端发起请求。服务端服用连接，完成客户端指定请求对应的操作，返回响应

3. 客户端断开链接。服务端断开链接，释放线程占有资源。

以上架构方案会有如下问题：

1. 客户端与服务端的连接与请求可能是很频繁的，比如采用 HTTP1.0 协议时，每一次连接与请求都是一一对应的。在高并发的背景下，过多同时间的请求会伴随着同样量级的连接，而这会对服务器开启与关闭线程/进程造成很大的处理压力。

2. 服务端的处理流程本质还是：连接->等待请求->处理->返回响应，的线性串行流程。当客户端建立连接而在请求前阻塞（可能是客户端自身，或者网络问题），服务端处理的线程也会等同阻塞，加之高并发的背景，这会导致服务端计算资源浪费、吞吐量下降。

针对以上两个问题，有如下解决方法。

池化技术

针对问题 1，服务端可以预先创建线程池，降低重复创建与关闭线程的额外开销。通过维护满足服务端承载处理能力的线程池，当有客户端连接请求时，可直接使用线程池中的线程处理。

额外，池化技术也是软件工程中常用的技术，除了上面提到的线程池，类似的 Golang 中协程池，与 MySQL、Redis 的连接池，各种业务场景中的对象池。都是采用类似的“空间换时间”的思想，预先执行消耗资源的初始化操作，而后在承载实际处理请求时服用各种资源。

非阻塞与主动感知

池化技术只能解决问题 1。但不能解决问题 2。问题 2 主要是串行处理会存在同步阻塞的问题。解决这个问题之前还需要一个铺垫：

1. 假设等待客户端请求（从服务端角度看，就是等待数据输入）与处理客户端请求可以分开两步完成。

那么解决这个问题可以有两种方法：

1. 化阻塞为非阻塞：数据输入阻塞时，服务端线程并不阻塞在这里，而是优先处理其他数据输入已准备好的请求。

2. 这样，服务端处理线程便可以不阻塞在等待客户端数据输入，而是一直保持在处理输入数据的状态。

3. 但这需要另一个线程通过不断轮询去感知到有哪些线程的数据输入是否准备好。

4. 考虑当需要轮询的线程增多时，完成一次轮询的时间会增长，这会间接导致已经准备好输入数据的线程得到下一次处理响应的延迟增长。

5. 化被动为主动：以上两种思路都是在服务端需要被动感知客户端输入数据是否准备好。如果能有第三方能主动通知服务端何时何对象需要被处理，那么服务端的处理效率将会得到极大增强。而这个第三方，就是操作系统提供给服务端的多路复用能力，即 IO 多路复用技术。

IO 多路复用

简单来说，IO 多路复用技术就是操作系统提供给服务端的一个外放能力。通过这个外放能力，服务端可以做到：

1. 服务端可以将需要感知的连接告诉操作系统。

2. 操作系统保证当这些连接数据输入已准备完成，便会通知到服务端。

3. 服务端能感知到输入数据准备完成的连接，就可以针对这些连接去执行对应的处理操作。

具体的 IO 多路复用技术在不同的系统中会有不同的实现，在这里就不赘述（我也还不懂）：

1. Linux 中，有 select/poll/epoll 三种实现

2. Windows 中，有 Windows Socket Asynchronous APIs、IOCP 等

Reactor 模式

以上 IO 多路复用技术已经能很好的解决最开始提出的问题。那么为了更符合直觉，将面向对象的编程思维引入进来，便创造了 Reactor/Dispatcher 模式。有一个解释：

1. 监听事件，并对其进行反应（React），而后将事件分发（Dispatch）到某个处理线程/进程中。

从对象的角度思考，Reactor 模式有如下两个对象：

1. Reactor：负责监听与分发事件（Event）。按照之前的例子，这里事件对应的就是连接与请求。

2. Handler：负责处理事件，执行事件对应的业务逻辑。如：连接事件，对应建立与客户端的连接；请求事件，对应执行客户端期望的获取/操作资源的逻辑。

从这两个对象可以看出，Handler 一般会处理较为复杂沉重的业务逻辑，而 Reactor 相对轻量。

Reactor 与 Handler 的实例数量可以根据不同场景分别水平扩展。简单的排列组合有以下几种情况：

1. 单 Reactor 单 Handler

2. 单 Reactor 多 Handler

3. 多 Reactor 单 Handler

4. 多 Reactor 多 Handler

其中，多 Reactor 单 Handler 使用场景不多（本来 Handler 处理效率就较差，还多设置 Reactor，单 Handler 会处理不来。）不赘述。实际情况与应用实例如下：

1. 单 Reactor 单 Handler：Redis

2. 单 Reactor 多 Handler

3. 多 Reactor 多 Handler：Netty、Kafka

下面重点关注 Kafka Broker 中的网络架构。

Broker 的网络架构

与 Reactor 模式的简单映射

Kafka Broker 与 Reactor 模式的简单对应，其有如下角色：

1. Reactor->Acceptor：负责建立并分发连接

2. Handler->Processor：负责监听读写事件并解析请求和响应，同时将请求分发给后续的工作线程。

新增的组件

Kafka Broker 为了应对超高并发，对 Processer 做了更细致的拆分。将监听读写事件与解析请求、响应拆开：

1. Processer：负责监听读写事件。

2. RequestHandler：负责解析请求。RequestHandler 将读取到的客户端请求字节封装成 Request 对象。便于后续组件的处理。（这里使用了池化技术，将多个 RequestHandler 封装在一个KafkaRequestHandlerPool线程池中）。

3. API 层：Kafka API 负责纯粹的消息处理逻辑。

核心处理流程

1. 启动：Broker 启动后，会根据服务端配置参数server.properties初始化上述三种组件线程：

2. Acceptor：根据listeners配置，默认监听 Broker 本机地址与 9092 端口，底层基于 Java NIO 监听 Socket 的连接事件OP_ACCEPT；

3. Processor：根据num.network.threads配置，默认 3 个，即一个 Acceptor 线程对应 3 个 Processor 线程；

4. RequestHandler：根据num.io.threads配置，默认 8 个，被封装在一个KafkaRequestHandlerPool线程池中。

5. 连接请求：当客户端发起建立连接请求时，Acceptor 会监听到该事件，然后完成连接的建立，并把建立好连接的 SocketChannel 通过 Round Robin 轮询的方式分配给各个 Processor 线程；

6. 等待读：Processor 线程会把接收到的 SocketChannel，缓存到自己内部的一个队列（ConcurrentLinkedQueue）中，等待 SocketChannel 收到读请求；

7. 解析请求数据：当 SocketChannel 监听读事件OP_READ发生时，每个 Processor 会通过底层的 NIO 组件读取请求字节，封装成 Request 对象，发送到RequestChannel组件中；

8. RequestChannel内部有一个缓存 Request 请求的全局队列（ArrayBlockingQueue），默认最多可以缓存 500 个请求，可通过参数queued.max.requests配置，同时有 N 个（N 为 Processor 线程的总数）缓存 Reponse 响应的队列（ArrayBlockingQueue）；

9. 转交请求：KafkaRequestHandlerPool线程池中的RequestHandler线程，会不断从RequestChannel中获取 Request 请求，交给 Kafka API 层进行处理；

10. 处理请求与转交响应：Kafka API 层完成消息处理后，会将结果封装成 Response 对象，并入队到 RequestChannel 内部响应队列中。

11. 发送响应：Processor 线程会对RequestChannel的响应队列中的 Response 对象进行处理，当它内部的 SocketChannel 监听到OP_WRITE写事件后，就会解析 Reponse，利用底层 NIO 组件响应给客户端。

写在后面

正如这篇文章的标题，本文聊得比较浅显与直观。更多偏向概念的建设、基本组件的功能与交互。关于源码的研读可以查看参考资料里（个人觉得写的还不错的博客）。后面会再考虑结合具体的例子进行深入的讨论。

参考资料

1. 深度剖析：Kafka 请求是如何处理? 看完这篇文章彻底懂了！
   

2. 如何深刻理解Reactor和Proactor？ - 知乎 (zhihu.com)
   

3. 这次答应我，一举拿下 I/O 多路复用！ (qq.com)
   

4. https://www.tpvlog.com/article/303

5. https://www.tpvlog.com/article/304

6. https://www.tpvlog.com/article/305

7. https://www.tpvlog.com/article/306