摘要：前 23 篇一直在讲 Producer 端。从这篇开始翻到 Broker 端——Kafka 是怎么监听 9092 端口、接受 Producer 连接、把连接分发给 IO 线程的？这是整个 Broker 网络层的起点。

关键词：Kafka Broker、Acceptor、SocketServer、Processor、NIO、网络线程模型

前情：从 Producer 翻到 Broker

前面讲了 23 篇 Producer 端——从 send() 初始化到内存回收、异常处理、超时机制。但有个问题一直没回答：Producer 发出去的 TCP 连接，Broker 这边是怎么接的？

Kafka Broker 的网络层不是简单的 ServerSocket 监听。它用了经典的 Reactor 模式——一个 Acceptor 接连接，多个 Processor 处理 IO。这篇从源码看这个模型是怎么启动的。

01 从 Kafka.main() 到 SocketServer

Broker 启动的入口是 Kafka.main()，最终走到 KafkaServer.startup()。网络层的初始化在 SocketServer 里：

// KafkaServer.startup() 中this.socketServer=newSocketServer(config,metrics,time);this.socketServer.startup();

SocketServer.startup() 做了三件事：

// SocketServer.startup()publicvoidstartup(){// 1. 创建 RequestChannel（请求队列，供业务线程消费）this.requestChannel=newRequestChannel(...);// 2. 创建 Acceptor 线程（每个监听器一个）for(Listenerlistener:listeners){Acceptoracceptor=newAcceptor(...);acceptorThread=newThread(acceptor,"kafka-network-thread-acceptor");acceptorThread.start();}// 3. 创建 N 个 Processor 线程for(inti=0;i<numNetworkThreads;i++){Processorprocessor=newProcessor(i,...);Threadthread=newThread(processor,"network-processor-"+i);thread.start();}}

关键：Acceptor 是每个监听器一个（listener 决定监听哪个端口/协议），Processor 是全局共享的 N 个。默认 num.network.threads=3。

02 为什么是 1 + N 模型

为什么 Acceptor 只有一个，Processor 有 N 个？因为 accept() 操作本身很快——内核完成三次握手后把连接放到 accept 队列，accept() 就是从队列取一个。真正耗时的是读请求、写响应。

如果用单线程：accept 慢的时候 IO 也跟着慢，IO 慢的时候 accept 也堵。

如果每个连接一个线程：连接数上去了线程爆炸。

Kafka 的解法：Acceptor 只负责 accept，不做任何 IO；Processor 用 NIO Selector 处理大量连接的读写。这就是经典的"单 Reactor 多线程"模型。

Acceptor 接受连接后，不是自己处理，而是轮询分发给 Processor：

// Acceptor.run() 中分发逻辑privatevolatileintnextProcessorIndex=0;privateProcessornextProcessor(){// round-robin 轮询intidx=nextProcessorIndex;nextProcessorIndex=(nextProcessorIndex+1)+processors.size();returnprocessors.get(idx);}

每个 Processor 有自己的 Selector 和一个 newConnections 队列。Acceptor 把新连接塞进队列，Processor 自己在 run() 循环里取出来、注册 OP_READ。

03 Acceptor.run() 核心循环

Acceptor 的核心就是一个 while 循环：

// Acceptor.run()publicvoidrun(){// 绑定端口serverSocketChannel=ServerSocketChannel.open();serverSocketChannel.bind(newInetSocketAddress(port));serverSocketChannel.register(selector,SelectionKey.OP_ACCEPT);while(isRunning){// 1. 等待 accept 事件，500ms 超时selector.select(500);Set<SelectionKey>keys=selector.selectedKeys();for(SelectionKeykey:keys){if(key.isAcceptable()){// 2. 接受新连接SocketChannelsocketChannel=serverSocketChannel.accept();// 3. 限流检查if(connectionLimitReached(socketChannel)){socketChannel.close();continue;}// 4. 轮询选 Processor，加入其队列Processorprocessor=nextProcessor();processor.accept(newConnection(...));}}keys.clear();}}

几个细节值得注意：

select(500ms) 不是 select()。为什么带超时？因为需要定期检查 isRunning 标志——如果 Broker 要关闭，不能无限阻塞在 select 上。500ms 超时后醒一次，看看是不是该退出了。

accept() 是非阻塞的。ServerSocketChannel 配置了 non-blocking，accept() 立即返回。如果 accept 队列里没有连接，返回 null。

限流检查在分发之前。max.connections 和 max.connections.per.ip 控制全局和单 IP 连接数。超限的连接直接 close——不进 Processor 队列。

04 一个容易忽略的设计：newConnections 队列

Acceptor 调 processor.accept()，不是直接把 SocketChannel 注册到 Processor 的 Selector 上，而是先塞进一个队列：

// Processor.accept()publicvoidaccept(Connectionconnection){newConnections.add(connection);wakeup();// 唤醒 Processor 的 selector}

为什么要队列？因为 Acceptor 和 Processor 是不同线程。Acceptor 不能直接操作 Processor 的 Selector（Selectable 不是线程安全的）。Processor 自己在 run() 循环里：

// Processor.run() 开头publicvoidrun(){// 把 newConnections 队列里的连接全部注册到 selectorconfigureNewConnections();// 然后才开始 selectselector.poll(pollTimeoutMs);}

这个设计和 Producer 端的 RecordAccumulator 思路一致——跨线程通信通过队列，不直接共享状态。

写在最后

Acceptor 线程虽然代码不长，但它决定了整个 Broker 网络层的骨架：一个线程接连接，N 个线程处理 IO，连接通过队列分发，Selector 各自独立。