为什么 Sender 是独立线程
前面讲 RecordAccumulator 时反复提到：业务线程只做 append（μs 级内存操作），网络 IO 全部交给 Sender。这个设计是 Kafka 高吞吐的基石。
如果业务线程自己发消息，一次网络 RTT 几 ms，100 个线程并发写就会有 100 条 Socket 连接，Broker 端连接数爆炸。Sender 单线程统一管理网络，连接数 = Broker 数，所有请求串行入队、按 Broker 分组批量发送——这是典型的"单线程 Reactor"模式，和 Netty、Redis 的思路一致。
01 Sender 线程的运行循环
Sender 是一个KafkaThread，在 KafkaProducer 构造函数末尾启动：
this.sender=newSender(logContext,client,this.metadata,this.accumulator,...);ThreadioThread=newKafkaThread("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-flight：batch 从 Deque 移除，
baseOffset占位（等 ACK 后填真实 offset） - 按 nodeId 分组：
Map<Integer, List<ProducerBatch>> - 如果遇到未知分区（元数据里没有 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 按错误码走三条路：
成功（错误码=0）：
batch.done(offset,null);// 完成 Future，业务线程拿到 offsetaccumulator.deallocate(batch);// 归还 BufferPool 内存块注意 deallocate 是在 Sender 线程做的，不是业务线程。业务线程的 Future 完成后只是被唤醒，内存释放由 Sender 负责——这避免了多线程并发操作 BufferPool 的 free 队列。
可重试失败（NotLeaderForPartition、RequestTimedOut 等）：
batch.reenqueue();// 塞回 Deque 队头，等下一轮 drain// 下次发送前等待 retry.backoff.ms（默认100ms）reenqueue 用的是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 做维度转换（分区→Broker），NetworkClient 做非阻塞 IO，InFlightRequests 做限流和超时，响应处理做回调和内存回收——四个组件各司其职，组合成一个高吞吐、低延迟、可重试的发送流水线。
