Kafka网络层第一道门：Acceptor线程是怎么启动的

摘要前 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() 到 SocketServerBroker 启动的入口是 Kafka.main()最终走到 KafkaServer.startup()。网络层的初始化在 SocketServer 里// KafkaServer.startup() 中this.socketServernewSocketServer(config,metrics,time);this.socketServer.startup();SocketServer.startup() 做了三件事// SocketServer.startup()publicvoidstartup(){// 1. 创建 RequestChannel请求队列供业务线程消费this.requestChannelnewRequestChannel(...);// 2. 创建 Acceptor 线程每个监听器一个for(Listenerlistener:listeners){AcceptoracceptornewAcceptor(...);acceptorThreadnewThread(acceptor,kafka-network-thread-acceptor);acceptorThread.start();}// 3. 创建 N 个 Processor 线程for(inti0;inumNetworkThreads;i){ProcessorprocessornewProcessor(i,...);ThreadthreadnewThread(processor,network-processor-i);thread.start();}}关键Acceptor 是每个监听器一个listener 决定监听哪个端口/协议Processor 是全局共享的 N 个。默认 num.network.threads3。02 为什么是 1 N 模型为什么 Acceptor 只有一个Processor 有 N 个因为 accept() 操作本身很快——内核完成三次握手后把连接放到 accept 队列accept() 就是从队列取一个。真正耗时的是读请求、写响应。如果用单线程accept 慢的时候 IO 也跟着慢IO 慢的时候 accept 也堵。如果每个连接一个线程连接数上去了线程爆炸。Kafka 的解法Acceptor 只负责 accept不做任何 IOProcessor 用 NIO Selector 处理大量连接的读写。这就是经典的单 Reactor 多线程模型。Acceptor 接受连接后不是自己处理而是轮询分发给 Processor// Acceptor.run() 中分发逻辑privatevolatileintnextProcessorIndex0;privateProcessornextProcessor(){// round-robin 轮询intidxnextProcessorIndex;nextProcessorIndex(nextProcessorIndex1)processors.size();returnprocessors.get(idx);}每个 Processor 有自己的 Selector 和一个 newConnections 队列。Acceptor 把新连接塞进队列Processor 自己在 run() 循环里取出来、注册 OP_READ。03 Acceptor.run() 核心循环Acceptor 的核心就是一个 while 循环// Acceptor.run()publicvoidrun(){// 绑定端口serverSocketChannelServerSocketChannel.open();serverSocketChannel.bind(newInetSocketAddress(port));serverSocketChannel.register(selector,SelectionKey.OP_ACCEPT);while(isRunning){// 1. 等待 accept 事件500ms 超时selector.select(500);SetSelectionKeykeysselector.selectedKeys();for(SelectionKeykey:keys){if(key.isAcceptable()){// 2. 接受新连接SocketChannelsocketChannelserverSocketChannel.accept();// 3. 限流检查if(connectionLimitReached(socketChannel)){socketChannel.close();continue;}// 4. 轮询选 Processor加入其队列ProcessorprocessornextProcessor();processor.accept(newConnection(...));}}keys.clear();}}几个细节值得注意select(500ms) 不是 select()。为什么带超时因为需要定期检查 isRunning 标志——如果 Broker 要关闭不能无限阻塞在 select 上。500ms 超时后醒一次看看是不是该退出了。accept() 是非阻塞的。ServerSocketChannel 配置了 non-blockingaccept() 立即返回。如果 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 的 SelectorSelectable 不是线程安全的。Processor 自己在 run() 循环里// Processor.run() 开头publicvoidrun(){// 把 newConnections 队列里的连接全部注册到 selectorconfigureNewConnections();// 然后才开始 selectselector.poll(pollTimeoutMs);}这个设计和 Producer 端的 RecordAccumulator 思路一致——跨线程通信通过队列不直接共享状态。写在最后Acceptor 线程虽然代码不长但它决定了整个 Broker 网络层的骨架一个线程接连接N 个线程处理 IO连接通过队列分发Selector 各自独立。