很多人以为producer.send(record)就是把消息发给 Kafka——一行代码一个网络请求。真实情况是send() 根本不发网络请求它只写内存。真正发请求的是另一条线程Sender而且是攒够了一批才发。理解这个设计就理解了 KafkaProducer 为什么吞吐高、为什么 linger.ms 调大一点延迟换吞吐、为什么 buffer.memory 不够时 send() 会阻塞。01 四个阶段全景一眼看完整条链路分四个阶段跨两个线程业务线程序列化 → 分区 → 写进 RecordAccumulator 的某个 DequeRecordAccumulator按 TopicPartition 分桶攒批满了或时间到了标记 readySender 线程轮询 ready 批次按节点合并成 ProduceRequestBroker写本地日志等 ISR 复制返回响应业务线程和 Sender 线程之间没有直接调用靠的是 RecordAccumulator 这个共享内存队列。业务线程往里写Sender 往外拉中间的同步靠 Deque 本身的线程安全和几个 Condition。这就是为什么 send() 能这么快——它只是把一条字节数组塞进一个双端队列纳秒级。真正的网络 IO 全在另一个线程里批量做。02 序列化与分区send() 里最先做的两件事doSend()方法里waitOnMetadata 拿到 leader 之后紧接着是两步// 简化后的 doSend 核心逻辑ProducerRecordK,VinterceptedonInterceptors(topic,record);byte[]serializedKeykeySerializer.serialize(topic,intercepted.key());byte[]serializedValuevalueSerializer.serialize(topic,intercepted.value());intpartitionpartitioner.partition(topic,intercepted.key(),serializedKey,intercepted.value(),serializedValue,cluster);// 找到或创建该分区的 DequetryAppendRecordAppendResultresultaccumulator.append(topicPartition,timestamp,serializedKey,serializedValue,...);序列化器很直接StringSerializer 就是String.getBytes(UTF_8)。真正要注意的是分区选好了这个消息后续就归这个 TopicPartition 管——它和同分区的消息共用一个 ProducerBatch。分区器是攒批的关键有 keyUtils.toPositive(Utils.murmur2(keyBytes)) % numPartitions——同 key 永远落同分区保证顺序无 key老版本轮询新版本2.4默认用StickyPartitioner——先随机粘一个分区攒满再切大幅提升攒批效率这里有个反直觉的点无 key 时用轮询反而会让每个分区都攒不满。100 条消息分给 10 个分区每个分区只有 10 条每个批次都半满就发了。StickyPartitioner 的做法是先把这 100 条全塞到一个分区攒满 16KB 再切下一个——单批次更大网络 IO 更少。03 RecordAccumulator攒批的核心这是整个 Producer 最关键的组件也是最容易讲错的。它不是一个队列是一个按分区分桶的 MapConcurrentMapTopicPartition, DequeProducerBatch batches每个 TopicPartition 对应一个双端队列队列里是 ProducerBatch。append 从队尾加Sender 从队头取。append 的逻辑简化privateRecordAppendResulttryAppend(longtimestamp,byte[]key,byte[]value,...){DequeProducerBatchdqdeque(tp);synchronized(dq){ProducerBatchlastdq.peekLast();if(last!null){RecordAppendResultresultlast.tryAppend(...);if(result!null)returnresult;// 最后一个 batch 还能塞}// 塞不下了新建一个 ProducerBatchByteBuffernewBufferfree.allocate(size,maxTimeToBlock);ProducerBatchbatchnewProducerBatch(tp,newBuffer,...);batch.tryAppend(...);dq.addLast(batch);returnnewRecordAppendResult(batch.futures(),...);}}三个关键动作看队尾那个 batch 还能不能塞——能塞就塞进去不新建塞不下了——向 BufferPool 申请一块新内存默认 16KB建新 batch加到队尾buffer 不够了——业务线程阻塞在free.allocate()等 max.block.ms默认 60 秒批次何时 ready不是业务线程塞完就发要满足三个条件之一batch.size 满了默认 16KBlinger.ms 到了默认 0常用 5~20msDeque 里还有其他待发送批次比如上一个失败在重试linger.ms0 意味着只要有消息就立刻标记 ready——这是低延迟模式。linger.ms10 意味着等 10ms 看能不能多攒点——用 10ms 延迟换更大的批次、更少的网络请求。04 Sender单线程拉批次发请求Sender 是个 Threadrun() 里循环调 runOnce()。核心三步// 简化后的 Sender.runOncelongnowtime.milliseconds();ClusterAndWaitTimewaitOnMetadata...;// 上一篇讲的// 1. 哪些节点有 ready 批次ReadyCheckResultreadyaccumulator.ready(now);// 2. 拉这些节点的 ready 批次MapNode,ListProducerBatchbatchesaccumulator.drain(ready.readyNodes,...);// 3. 按节点拼成请求发出去for(Nodenode:batches.keySet()){ClientRequestreqclient.newProduceRequest(node,batches.get(node));inFlightRequests.add(req);}// 4. poll 等待响应selector.poll(pollTimeout);drain() 是攒批的最终目的它把同一个 leader 节点上所有 TopicPartition 的 ready 批次合并成一个 ProduceRequest。一个请求里可能包含 topic-A-0、topic-A-1、topic-B-0 三个分区的数据——一次 TCP 往返带过去。InFlightRequests限制每个节点最多 5 个未完成请求max.in.flight.requests.per.connection。不是越多越好——太多会乱序太少会浪费带宽。acksall 时建议保持默认 5enable.idempotencetrue 时这个值必须 ≤5。响应回来后成功completeFuture业务线程那边拿到的 FutureRecordMetadata 终于完成回调 onCompletion 触发可重试错误leader 切换、网络闪断批次放回 Deque 队头等下一轮再发不可重试错误序列化错、鉴权失败completeFuture 带异常回调拿到错误05 关键配置与 JMX 判断配置默认值作用batch.size16384单 batch 最大字节不是单消息大小linger.ms0攒批等待时间调大换吞吐buffer.memory33554432总缓冲 32MB满了 send 阻塞compression.typenonebatch 层压缩lz4/zstd 推荐max.request.size1048576单请求最大 1MBacksall0/1/all 三种可靠性级别JMX 上判断 Producer 是不是卡在攒批buffer-available-bytes持续为 0缓冲耗尽send 在阻塞调大 buffer.memory 或排查慢 Brokerbatch-size-avg很低比如平均 2KBlinger.ms0 或消息太少攒批没生效record-queue-time-avg高消息在 Deque 里排太久Sender 发得慢查 Broker 负载record-send-rate和record-error-rate发送速率和错误率对比看06 排查清单send() 不抛异常但 Broker 没收到查 buffer.memory 是不是满了或 Sender 线程挂了看 record-error-rate 和 JMX吞吐上不去linger.ms0 导致批次太小调大到 5~20ms开 compression.typelz4send() 偶发阻塞几百毫秒正常BufferPool 在等旧 batch 归还持续阻塞就查 Broker同 key 消息乱序max.in.flight.requests.per.connection 1 且 retries 0 时会乱序设为 1 或开 enable.idempotenceOOMbuffer.memory 设太大或消息太大超过 max.request.size——后者会直接抛异常不会 OOM写在最后Producer 核心流程的设计精髓就一句话把每条消息一个请求改成攒一批发一个请求。业务线程只管往内存里写Sender 线程统一拉、统一合并、统一发。两个线程之间靠 RecordAccumulator 解耦——业务线程永远不会因为网络慢而被拖住最多就是缓冲满了阻塞一下。这个设计带来三个直接好处吞吐高批次大、网络少、延迟可控linger.ms 调、削峰填谷业务高峰先堆内存Sender 按能力慢慢发。代价就是多了一层内存缓冲——故障排查时要理解消息在路上可能在三个地方业务线程正在 append、Deque 里等着、in-flight 请求里没响应。