摘要：前 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 = new SocketServer(config, metrics, time);
this.socketServer.startup();
SocketServer.startup() 做了三件事：
// SocketServer.startup()
public void startup() {
// 1. 创建 RequestChannel（请求队列，供业务线程消费）
this.requestChannel = new RequestChannel(...);
// 2. 创建 Acceptor 线程（每个监听器一个）
for (Listener listener : listeners) {
Acceptor acceptor = new Acceptor(...);
acceptorThread = new Thread(acceptor, "kafka-network-thread-acceptor");
acceptorThread.start();
}
// 3. 创建 N 个 Processor 线程
for (int i = 0; i < numNetworkThreads; i++) {
Processor processor = new Processor(i, ...);
Thread thread = new Thread(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() 中分发逻辑
private volatile int nextProcessorIndex = 0;
private Processor nextProcessor() {
// round-robin 轮询
int idx = nextProcessorIndex;
nextProcessorIndex = (nextProcessorIndex + 1) + processors.size();
return processors.get(idx);
}
每个 Processor 有自己的 Selector 和一个 newConnections 队列。Acceptor 把新连接塞进队列，Processor 自己在 run() 循环里取出来、注册 OP_READ。
03 Acceptor.run() 核心循环
Acceptor 的核心就是一个 while 循环：
// Acceptor.run()
public void run() {
// 绑定端口
serverSocketChannel = ServerSocketChannel.open();
serverSocketChannel.bind(new InetSocketAddress(port));
serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
while (isRunning) {
// 1. 等待 accept 事件，500ms 超时
selector.select(500);
Set<SelectionKey> keys = selector.selectedKeys();
for (SelectionKey key : keys) {
if (key.isAcceptable()) {
// 2. 接受新连接
SocketChannel socketChannel = serverSocketChannel.accept();
// 3. 限流检查
if (connectionLimitReached(socketChannel)) {
socketChannel.close();
continue;
}
// 4. 轮询选 Processor，加入其队列
Processor processor = nextProcessor();
processor.accept(new Connection(...));
}
}
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()
public void accept(Connection connection) {
newConnections.add(connection);
wakeup(); // 唤醒 Processor 的 selector
}
为什么要队列？因为 Acceptor 和 Processor 是不同线程。Acceptor 不能直接操作 Processor 的 Selector（Selectable 不是线程安全的）。Processor 自己在 run() 循环里：
// Processor.run() 开头
public void run() {
// 把 newConnections 队列里的连接全部注册到 selector
configureNewConnections();
// 然后才开始 select
selector.poll(pollTimeoutMs);
}
这个设计和 Producer 端的 RecordAccumulator 思路一致——跨线程通信通过队列，不直接共享状态。
写在最后
Acceptor 线程虽然代码不长，但它决定了整个 Broker 网络层的骨架：一个线程接连接，N 个线程处理 IO，连接通过队列分发，Selector 各自独立。
转载自 CSDN-专业IT技术社区
原文链接：https://blog.csdn.net/zmj_0817/article/details/166007620
