4. 批量发送原理（上）：batch.size 与 linger.ms 参数详解、消息在 RecordAccumulator 中的缓冲与打包过程

Kafka 高性能的基石之一在于其批量发送机制。生产者不会每发一条消息就立即发起网络请求，而是先将消息暂存在内存缓冲区中，待满足特定条件后，再将一批消息打包发送。本章将深入剖析这一过程的核心参数与内部数据结构。

4.1 核心参数：batch.size 与 linger.ms

批量发送的行为由两个关键参数共同控制：batch.size 决定“空间阈值”，linger.ms 决定“时间阈值”。任何一个条件先满足，都会触发当前批次的发送。

4.1.1 batch.size 详解

• 单位：字节（bytes）。
• 并非硬性限制：单个批次的实际大小可以略微超过 batch.size，因为单条消息如果本身就大于 batch.size，它仍然会被单独作为一个批次发送。
• 内存分配：Kafka 生产者会为每个分区预先分配固定大小的 BufferPool 内存块。批次大小直接影响内存使用效率。
• 调优建议：如果消息平均大小为 1 KB，将 batch.size 设为 32 KB，则每个批次可容纳约 32 条消息，能有效减少网络请求次数。

4.1.2 linger.ms 详解

• 单位：毫秒（ms）。
• 与 batch.size 协同：linger.ms 提供了一个“等待窗口”。即使当前批次未满，只要等待时间到达，也会立即发送。
• 对延迟的影响：增大 linger.ms 会增加消息的端到端延迟（因为消息在缓冲区等待），但能显著提升吞吐量。
• 典型配置：对于日志收集等吞吐优先场景，建议 linger.ms=10~50；对于实时交易等延迟敏感场景，建议保持默认值 0 或设为 1~2 ms。

4.2 消息在 RecordAccumulator 中的缓冲与打包过程

RecordAccumulator 是 Kafka 生产者客户端的核心缓冲组件，它负责接收来自 KafkaProducer.send() 的消息，并将其组织成可发送的批次。整个过程分为三个关键阶段：追加、打包、唤醒。

4.2.1 阶段一：消息追加（Append）

1. 计算目标分区：根据消息的 key 和分区策略，确定消息属于哪个 TopicPartition。
2. 获取队列：从 RecordAccumulator 的 batches 映射中获取该分区对应的 Deque<ProducerBatch>。
3. 尝试追加：获取队列尾部的最后一个 ProducerBatch（即当前正在写入的批次），尝试将消息追加进去。
4. 成功或新建：
   • 如果追加成功（批次未满且内存足够），则直接返回 RecordAppendResult。
   • 如果追加失败（批次已满或不存在），则从 BufferPool 申请一块新的内存，创建一个新的 ProducerBatch，将消息追加到新批次中，并将新批次放入队列尾部。

4.2.2 阶段二：批次打包与发送准备

当消息被追加到 ProducerBatch 后，该批次并不会立即发送。它会在队列中等待，直到满足以下任一条件：

• 条件 A（空间满）：批次累积的数据大小达到 batch.size。
• 条件 B（时间到）：从批次创建开始，等待时间超过 linger.ms。

一旦条件满足，Sender 线程（后台 I/O 线程）会在其运行循环中，从 RecordAccumulator 中拉取所有符合条件的 ProducerBatch，并将它们封装成 ProducerRequest，准备发往对应的 Broker。

4.2.3 阶段三：Sender 线程唤醒与 Drain

Sender 线程会持续轮询 RecordAccumulator。当有新的批次准备好发送时，KafkaProducer.send() 方法会通过 wakeup() 机制唤醒 Sender 线程。Sender 线程执行 Drain（排水） 操作：

• 遍历所有分区队列，检查每个队列头部的 ProducerBatch。
• 如果该批次已满足发送条件（满或超时），则将其从队列中移除，并加入待发送列表。
• 将待发送列表按 Node（Broker 节点）分组，构建 ClientRequest，最终通过 NetworkClient 发送到网络。

4.3 批量发送流程总览

下图展示了从 send() 调用到消息最终被封装成网络请求的完整流程，重点突出了 RecordAccumulator 的内部状态变化。

4.4 参数调优对照表

在实际压测与调优中，batch.size 和 linger.ms 需要根据消息体大小和目标吞吐量进行组合调整。下表给出了不同场景下的推荐配置与预期效果：