为什么 Sender 是独立线程前面讲 RecordAccumulator 时反复提到业务线程只做 appendμs 级内存操作网络 IO 全部交给 Sender。这个设计是 Kafka 高吞吐的基石。如果业务线程自己发消息一次网络 RTT 几 ms100 个线程并发写就会有 100 条 Socket 连接Broker 端连接数爆炸。Sender 单线程统一管理网络连接数 Broker 数所有请求串行入队、按 Broker 分组批量发送——这是典型的单线程 Reactor模式和 Netty、Redis 的思路一致。01 Sender 线程的运行循环Sender 是一个KafkaThread在 KafkaProducer 构造函数末尾启动this.sendernewSender(logContext,client,this.metadata,this.accumulator,...);ThreadioThreadnewKafkaThread(kafka-producer-network-thread | clientId,sender,true);ioThread.start();启动后进入无限循环while (running) { runOnce() }每轮 runOnce 做四件事第一步 drain()从 RecordAccumulator 拉取可发送的批次按 Broker 分组。这一步是内存 → 网络的交接点——drain 出来的批次从 Deque 移除标记为 in-flight业务线程不再能修改。第二步 sendProduceRequests()把每个 Broker 的批次列表构建成一个ProduceRequest调用NetworkClient.send()封装成ClientRequest入队。注意这一步还没真正写 Socket只是把请求放进 NetworkClient 的待发送队列。第三步 client.poll()真正的网络 IO。Selector 用 NIO 非阻塞模式把待发送请求写入 Socket同时检查是否有响应可读。poll 的超时时间是min(pollTimeout, 最近的退避时间)——有重试请求在退避时poll 会精确等到退避结束再返回不空转。第四步 handleResponse()处理 poll 返回的响应。成功 → 回调业务的 Future 释放 BufferPool 内存可重试失败 → 批次重新塞回 Deque 队头等下一轮不可重试失败 → 回调异常。02 drain从按分区到按 Broker的转换drain 是 Sender 最核心的逻辑也是理解 Kafka 高吞吐的关键。业务线程按分区写每个 TopicPartition 一个 Deque但网络是按 Broker 发的——drain 做的就是这个维度转换。drain 的执行步骤遍历readyNodes有 ready 批次的 Broker 集合只处理这些节点对每个节点遍历它负责的所有分区从 Deque 队头取 ready 状态的 batch限流检查该节点的 in-flight 请求数 max.in.flight.requests.per.connection默认5才继续大小限制累计字节数 ≤max.request.size默认1MB超过就停标记 in-flightbatch 从 Deque 移除baseOffset占位等 ACK 后填真实 offset按 nodeId 分组MapInteger, ListProducerBatch如果遇到未知分区元数据里没有 Leader触发metadata.requestUpdate()输出是按 Broker 分组的批次集合——3 个 Broker 的集群drain 完可能得到 Node-0 有 2 个批次、Node-1 有 1 个批次、Node-2 没有。然后每个非空节点构建一个 ProduceRequest。这里有个容易忽略的细节drain 是按节点遍历的不是按分区遍历的。先确定哪些 Broker 有 ready 批次再去拉对应分区的批次。这样做的好处是 in-flight 限流是节点级的不会因为某个分区流量大就把整个节点的配额占满。03 NetworkClient非阻塞发送与 InFlight 跟踪sendProduceRequests 构建完 ProduceRequest 后调用NetworkClient.send(clientRequest, now)。这一步不写 Socket只做两件事把请求放进InFlightRequests每个节点一个双端队列计数1把请求关联到对应的KafkaChannel等下一次 poll 时写入Selector 的 poll 做什么pollSelectionKeys()遍历所有已连接的 KafkaChannel对写就绪的通道调用channel.write()把请求字节发出去对读就绪的通道调用channel.read()读响应completedSends写完的请求记录下来completedReceives读完的响应记录下来disconnected断开的连接记录下来poll 返回后NetworkClient 遍历completedReceives把每个响应解析成ClientResponse从 InFlightRequests 中找到对应的请求按 correlationId 匹配然后调用request.callback().onComplete(response)。InFlightRequests 的作用限流每个节点的队列长度 ≤ max.in.flight超过就不再发超时检测poll 时遍历检查nowMs - createdTimeMs requestTimeoutMs超时的请求断开连接避免收到脏响应按失败处理顺序保证同节点的请求按 correlationId 递增响应按序匹配04 响应处理成功、重试、失败三条路收到 ProduceResponse 后Sender 按错误码走三条路成功错误码0batch.done(offset,null);// 完成 Future业务线程拿到 offsetaccumulator.deallocate(batch);// 归还 BufferPool 内存块注意 deallocate 是在 Sender 线程做的不是业务线程。业务线程的 Future 完成后只是被唤醒内存释放由 Sender 负责——这避免了多线程并发操作 BufferPool 的 free 队列。可重试失败NotLeaderForPartition、RequestTimedOut 等batch.reenqueue();// 塞回 Deque 队头等下一轮 drain// 下次发送前等待 retry.backoff.ms默认100msreenqueue 用的是addFirst()——重传的批次已经 ready应该优先发不能排在还在攒的新批次后面。这就是前面讲 Deque 双端设计时提到的重传插队头。重试有总超时限制delivery.timeout.ms默认120s从 append 开始算包含所有重试退避时间。超过这个时间还没成功就走失败分支。不可重试失败MessageTooLarge、InvalidTopicException 等batch.done(-1,exception);// Future 抛异常accumulator.deallocate(batch);// 内存照样释放不泄漏05 超时处理request.timeout.ms vs delivery.timeout.ms这两个参数容易混淆明确区分request.timeout.ms默认30s单次请求的超时。从请求发出到收到响应的时间超过就断开连接、按失败处理。可重试的会重试。delivery.timeout.ms默认120s从 append 到最终成功或失败的总时间。包含所有重试退避。超过这个时间即使还能重试也放弃直接回调失败。约束关系delivery.timeout.ms ≥ linger.ms request.timeout.ms否则配置校验会报错。超时检测在 poll 里做每轮 poll 遍历 InFlightRequests 里的所有请求检查是否超时。超时的请求会触发连接断开channel.disconnect()因为不确定对端是否已经收到请求保留连接可能收到脏响应。写在最后Sender 线程的设计精髓可以浓缩成一句话业务线程管内存Sender 线程管网络两者通过 RecordAccumulator 解耦。drain 做维度转换分区→BrokerNetworkClient 做非阻塞 IOInFlightRequests 做限流和超时响应处理做回调和内存回收——四个组件各司其职组合成一个高吞吐、低延迟、可重试的发送流水线。
