企业官网建设流程全解析

很多人以为producer.send(record)就是"把消息发给 Kafka"——一行代码一个网络请求。真实情况是：send() 根本不发网络请求，它只写内存。真正发请求的是另一条线程（Sender），而且是攒够了一批才发。

理解这个设计，就理解了 KafkaProducer 为什么吞吐高、为什么 linger.ms 调大一点延迟换吞吐、为什么 buffer.memory 不够时 send() 会阻塞。

01 四个阶段：全景一眼看完

整条链路分四个阶段，跨两个线程：

• 业务线程：序列化 → 分区 → 写进 RecordAccumulator 的某个 Deque
• RecordAccumulator：按 TopicPartition 分桶攒批，满了或时间到了标记 ready
• Sender 线程：轮询 ready 批次，按节点合并成 ProduceRequest
• Broker：写本地日志，等 ISR 复制，返回响应

业务线程和 Sender 线程之间没有直接调用，靠的是 RecordAccumulator 这个共享内存队列。业务线程往里写，Sender 往外拉，中间的同步靠 Deque 本身的线程安全和几个 Condition。

这就是为什么 send() 能这么快——它只是把一条字节数组塞进一个双端队列，纳秒级。真正的网络 IO 全在另一个线程里批量做。

02 序列化与分区：send() 里最先做的两件事

doSend()方法里，waitOnMetadata 拿到 leader 之后，紧接着是两步：

// 简化后的 doSend 核心逻辑ProducerRecord<K,V>intercepted=onInterceptors(topic,record);byte[]serializedKey=keySerializer.serialize(topic,intercepted.key());byte[]serializedValue=valueSerializer.serialize(topic,intercepted.value());intpartition=partitioner.partition(topic,intercepted.key(),serializedKey,intercepted.value(),serializedValue,cluster);// 找到或创建该分区的 Deque，tryAppendRecordAppendResultresult=accumulator.append(topicPartition,timestamp,serializedKey,serializedValue,...);

序列化器很直接，StringSerializer 就是String.getBytes(UTF_8)。真正要注意的是：分区选好了，这个消息后续就归这个 TopicPartition 管——它和同分区的消息共用一个 ProducerBatch。

分区器是攒批的关键：

• 有 key：Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions——同 key 永远落同分区，保证顺序
• 无 key：老版本轮询，新版本（2.4+）默认用StickyPartitioner——先随机粘一个分区攒满再切，大幅提升攒批效率

这里有个反直觉的点：无 key 时用轮询反而会让每个分区都攒不满。100 条消息分给 10 个分区，每个分区只有 10 条，每个批次都半满就发了。StickyPartitioner 的做法是先把这 100 条全塞到一个分区，攒满 16KB 再切下一个——单批次更大，网络 IO 更少。

03 RecordAccumulator：攒批的核心

这是整个 Producer 最关键的组件，也是最容易讲错的。它不是一个队列，是一个按分区分桶的 Map：

ConcurrentMap<TopicPartition, Deque<ProducerBatch>> batches

每个 TopicPartition 对应一个双端队列，队列里是 ProducerBatch。append 从队尾加，Sender 从队头取。

append 的逻辑（简化）：

privateRecordAppendResulttryAppend(longtimestamp,byte[]key,byte[]value,...){Deque<ProducerBatch>dq=deque(tp);synchronized(dq){ProducerBatchlast=dq.peekLast();if(last!=null){RecordAppendResultresult=last.tryAppend(...);if(result!=null)returnresult;// 最后一个 batch 还能塞}// 塞不下了，新建一个 ProducerBatchByteBuffernewBuffer=free.allocate(size,maxTimeToBlock);ProducerBatchbatch=newProducerBatch(tp,newBuffer,...);batch.tryAppend(...);dq.addLast(batch);returnnewRecordAppendResult(batch.futures(),...);}}

三个关键动作：

1. 看队尾那个 batch 还能不能塞——能塞就塞进去，不新建
2. 塞不下了——向 BufferPool 申请一块新内存（默认 16KB），建新 batch，加到队尾
3. buffer 不够了——业务线程阻塞在free.allocate()，等 max.block.ms（默认 60 秒）

批次何时 ready？不是"业务线程塞完就发"，要满足三个条件之一：

• batch.size 满了（默认 16KB）
• linger.ms 到了（默认 0，常用 5~20ms）
• Deque 里还有其他待发送批次（比如上一个失败在重试）

linger.ms=0 意味着"只要有消息就立刻标记 ready"——这是低延迟模式。linger.ms=10 意味着"等 10ms 看能不能多攒点"——用 10ms 延迟换更大的批次、更少的网络请求。

04 Sender：单线程拉批次发请求

Sender 是个 Thread，run() 里循环调 runOnce()。核心三步：

// 简化后的 Sender.runOncelongnow=time.milliseconds();ClusterAndWaitTimewaitOnMetadata=...;// 上一篇讲的// 1. 哪些节点有 ready 批次ReadyCheckResultready=accumulator.ready(now);// 2. 拉这些节点的 ready 批次Map<Node,List<ProducerBatch>>batches=accumulator.drain(ready.readyNodes,...);// 3. 按节点拼成请求发出去for(Nodenode:batches.keySet()){ClientRequestreq=client.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）。不是越多越好——太多会乱序，太少会浪费带宽。acks=all 时建议保持默认 5，enable.idempotence=true 时这个值必须 ≤5。

响应回来后：

• 成功：completeFuture，业务线程那边拿到的 FutureRecordMetadata 终于完成，回调 onCompletion 触发
• 可重试错误（leader 切换、网络闪断）：批次放回 Deque 队头，等下一轮再发
• 不可重试错误（序列化错、鉴权失败）：completeFuture 带异常，回调拿到错误

05 关键配置与 JMX 判断

JMX 上判断 Producer 是不是卡在攒批：

• buffer-available-bytes持续为 0：缓冲耗尽，send 在阻塞，调大 buffer.memory 或排查慢 Broker
• batch-size-avg很低（比如平均 2KB）：linger.ms=0 或消息太少，攒批没生效
• record-queue-time-avg高：消息在 Deque 里排太久，Sender 发得慢，查 Broker 负载
• record-send-rate和record-error-rate：发送速率和错误率对比看

06 排查清单

1. send() 不抛异常但 Broker 没收到：查 buffer.memory 是不是满了，或 Sender 线程挂了（看 record-error-rate 和 JMX）
2. 吞吐上不去：linger.ms=0 导致批次太小，调大到 5~20ms；开 compression.type=lz4
3. send() 偶发阻塞几百毫秒：正常，BufferPool 在等旧 batch 归还；持续阻塞就查 Broker
4. 同 key 消息乱序：max.in.flight.requests.per.connection > 1 且 retries > 0 时会乱序，设为 1 或开 enable.idempotence
5. OOM：buffer.memory 设太大，或消息太大超过 max.request.size——后者会直接抛异常，不会 OOM

写在最后

Producer 核心流程的设计精髓就一句话：把"每条消息一个请求"改成"攒一批发一个请求"。业务线程只管往内存里写，Sender 线程统一拉、统一合并、统一发。两个线程之间靠 RecordAccumulator 解耦——业务线程永远不会因为网络慢而被拖住，最多就是缓冲满了阻塞一下。

这个设计带来三个直接好处：吞吐高（批次大、网络少）、延迟可控（linger.ms 调）、削峰填谷（业务高峰先堆内存，Sender 按能力慢慢发）。代价就是多了一层内存缓冲——故障排查时要理解消息"在路上"可能在三个地方：业务线程正在 append、Deque 里等着、in-flight 请求里没响应。