摘要上一篇讲了 Processor 怎么从网络读字节、怎么解析协议。但 poll() 返回的 completedReceives 列表里到底是什么Processor 遍历这个列表时做了哪些判断mute 限流机制是怎么回事为什么 KSelector 要在底层帮你过滤半包这篇把 completedReceives 的处理细节讲透。关键词Kafka Processor、completedReceives、KChannel、mute限流、RequestChannel、背压、KSelector前情poll() 返回了什么上一篇讲了 selector.poll() 会阻塞直到有连接可读。但 poll() 返回之后Processor 拿到的不是一堆裸 SocketChannel而是一个叫 completedReceives 的列表。这个列表里的每个元素都是已经读完了完整请求的连接——半包的不在里面。Selector 底层已经帮你判断过 ByteBuffer 够不够解析了。这是 Kafka NIO 设计里最容易被忽略的一层KSelector 不只是 Selector 的薄封装它在底层做了协议层面的判断——ByteBuffer 里有多少字节、够不够一条完整消息全在 KSelector 内部处理。Processor 拿到的已经是成品。01 completedReceives 全景一次 poll 可能返回多个就绪连接。比如同时有 Produce、Fetch、Metadata 请求到达。Processor 遍历这个列表逐个处理// Processor.processCompletedReceives()publicvoidprocessCompletedReceives(){for(KChannelchannel:selector.completedReceives()){// 1. 检查是否被静音if(channel.isMute())continue;// 2. 取出已准备好的请求NetworkReceivereceivechannel.fetchPreppedReceive();// 3. 解析 RequestHeaderRequestHeaderheaderRequestHeader.parse(receive.buffer());// 4. 按 apiKey 解析 bodyRequestrequestbuildRequest(header,receive);// 5. 送入 RequestChannelrequestChannel.sendRequest(request);}}注意半包连接不会出现在 completedReceives 里。KSelector 内部已经判断了——ByteBuffer 里够不够一条完整消息。不够就继续等够了才放进 completedReceives。那 KSelector 是怎么判断的核心在这个方法// KSelector.read() 内部privatevoidread(KChannelchannel)throwsIOException{intreadchannel.read();if(read0){// 连接断开}elseif(read0){// 有数据读进来了NetworkReceivereceivechannel.incompleteReceive();if(receivenull){receivenewNetworkReceive(channel);channel.setIncompleteReceive(receive);}// 判断是否读完了一条完整消息if(receive.complete()){channel.setPreppedReceive(receive);channel.clearIncompleteReceive();completedReceives.add(channel);// 放进来}}}关键就是receive.complete()——ByteBuffer 里的字节够不够一条完整消息。不够就留在 incompleteReceive 里够了才放进 completedReceives。02 KChannel 是什么你可能注意到了Processor 操作的不是裸 SocketChannel是 KChannel。KChannel 是 Kafka 对 SocketChannel 的包装附加了这些东西publicclassKChannel{privatefinalStringid;// 连接唯一标识privatefinalSocketChannelsocketChannel;// NIO原始通道privatefinalByteBufferrecvBuffer;// 读缓冲privatefinalByteBuffersendBuffer;// 写缓冲privateNetworkReceiveincompleteReceive;// 半包暂存privateNetworkReceivepreppedReceive;// 已解析length的完整请求privatebooleanisMute;// 是否静音privateListNetworkSendsends;// 待发送的响应队列// ... SASL认证信息}为什么要包一层因为裸 SocketChannel 只有 IO 操作没有连接状态。Kafka 需要记录这个连接是不是被限流了、认证状态是什么、ByteBuffer 里攒了多少数据、有多少个响应待发送。这些都放在 KChannel 里。incompleteReceive 和 preppedReceive 的区别是关键incompleteReceive半包ByteBuffer 里数据不够一条完整消息preppedReceive完整消息已经够解析了等 Processor 来取Processor 永远只从 preppedReceive 取——半包的它根本看不见。处理流程五步取出 KChannel → 检查 mute → 取出 preppedReceive → 解析 header → 组装 Request 入队。03 mute 限流机制这是最容易忽略的设计。如果 RequestChannel 的队列满了queued.max.requests 默认 500新请求就进不去了。Kafka 的做法不是阻塞 Processor而是mute 这个连接// RequestChannel 满了的判断publicbooleanhasStartedDequeuing(){returnstartDequeueTimeMs0;}// Processor 里的逻辑if(requestChannel.hasStartedDequeuing()isFull()){for(KChannelchannel:selector.channels()){if(!channel.isMute()){channel.mute();// 静音不再读这个连接的数据muted.add(channel);}}}mute 之后Selector 不会再对这个连接触发 OP_READ。即使数据到了也不会读进来。等业务线程消费了一些请求队列有空间了Processor 再 unmute// 队列有空位了if(!requestChannel.hasStartedDequeuing()||!isFull()){for(KChannelchannel:muted){channel.unmute();// 恢复读取}muted.clear();}这个设计叫背压——下游处理不过来上游就别再读了。不是压垮自己而是优雅地减速。为什么不在 Processor 线程里阻塞因为 Processor 线程要同时管所有连接。如果一个连接阻塞了其他连接都没法处理。mute 是更优雅的做法不阻塞线程只是暂时不读这个连接。04 批量处理的效率一次 poll 可能返回几十个就绪连接。Processor 在一个循环里全部处理完不会为每个连接单独 poll。为什么这样快因为 poll 本身是一次系统调用批量处理 N 个连接只需要一次 poll。如果每个连接单独 poll系统调用开销就把性能吃光了。这和 Linux epoll 的水平触发LT模式是配套的——一次 poll 返回所有就绪连接应用层批量处理。Kafka 用的是 LT 还是 ET答案是 LT。这意味着即使你一次没处理完所有就绪连接下次 poll 还会通知你。这也是为什么 mute 机制要显式控制——如果是 ET 模式mute 了就再也收不到通知了。05 实战什么场景下 mute 会触发线上如果遇到 mute 频繁触发通常是因为业务线程池处理不过来。常见原因RequestHandler 线程数太少默认 8消费速度跟不上磁盘 IO 慢写日志阻塞了 RequestHandler某个分区的请求处理特别慢拖慢了整个队列排查时看request-handler-pct这个指标——如果经常接近 100%说明业务线程已经饱和了该加线程数或者优化慢查询了。写在最后completedReceives 的处理核心就是三步遍历列表、跳过 mute 的连接、解析入队。mute 机制是 Kafka 网络层背压的关键——队列满了就静音队列空了就恢复优雅不丢请求。理解了 KSelector 在底层做的半包判断、KChannel 的状态包装、mute 背压机制你就理解了 Kafka 网络层怎么在一个线程里同时管成百上千个连接。