Kafka 消息的发送过程简单介绍一下？
结论：
send()是异步的，消息先在客户端缓冲、再由独立线程批量发出。完整链路是：拦截器（Interceptor）→ 序列化（Serializer）→ 分区器（Partitioner）→ 记录累加器（RecordAccumulator）→ 发送线程（Sender）按 Broker 聚合批次 → 网络客户端（NetworkClient）发出 → Broker 响应后回调Callback。主线程只负责"放进缓冲区"，真正的网络 IO 在 Sender 线程。
一、发送链路
main 线程 sender 线程
┌──────────────┐ ┌──────────────┐ ┌──────────────────┐
│ 1 Interceptor│──>│ 2 Serializer │──>│ 3 Partitioner │
└──────────────┘ └──────────────┘ └────────┬─────────┘
│ 选定分区
┌─────────▼──────────┐
│ 4 RecordAccumulator│ 每个分区一个双端队列
│ batch.size 攒批 │ 按 Broker 聚合
└─────────┬──────────┘
│ 缓冲满 / linger 到期
┌─────────▼──────────┐
│ 5 Sender 线程 │──> 6 NetworkClient
└─────────┬──────────┘ │
│ 响应 ▼
┌─────────▼──────────┐ ┌──────────┐
│ 7 Callback / Future│<───│ Broker │
└────────────────────┘ └──────────┘
二、各阶段职责
| 阶段 | 组件 | 关键配置 | 说明 |
|---|---|---|---|
| 拦截 | ProducerInterceptor | 无 | 可修改消息、统计耗时，注意别做重活 |
| 序列化 | Serializer | key.serializer / value.serializer | 失败直接抛 SerializationException |
| 分区 | Partitioner | partitioner.class | 有 key 用 murmur2(key) % 分区数，无 key 走粘性分区 |
| 缓冲 | RecordAccumulator | buffer.memory、batch.size | 按分区维护队列，满时阻塞 max.block.ms |
| 发送 | Sender | linger.ms、max.in.flight | 单线程轮询，按 Broker 聚合请求 |
| 网络 | NetworkClient | request.timeout.ms | 维护连接、在途请求、重试 |
| 回调 | Callback | 无 | 成功/失败都在 Sender 线程执行，不能阻塞 |
三、缓冲与批量
buffer.memory=33554432 # 生产者总缓冲 32MB，满了 send 阻塞
batch.size=16384 # 单分区单批次 16KB
linger.ms=5 # 最多等 5ms 让批次变大
max.block.ms=60000 # 缓冲满时最多阻塞 60s，超时抛异常
一次 send() 的行为分三种情况：
- 目标分区已有批次且未满 → 追加进去，立即返回；
- 批次已满 → 关闭当前批次，等 Sender 取走；
- 缓冲总量超限 → 阻塞到
max.block.ms，仍未释放就抛TimeoutException。
四、代码：异步与同步
// 异步：主线程不等待，回调里处理结果
producer.send(record, (metadata, e) -> {
if (e != null) {
log.error("send failed, key={}", record.key(), e);
// 落本地补偿表，由定时任务重投
}
});
// 同步：get() 会阻塞到 Broker 响应，吞吐低，只用于关键少量消息
try {
RecordMetadata md = producer.send(record).get(3, TimeUnit.SECONDS);
log.info("offset={} partition={}", md.offset(), md.partition());
} catch (Exception e) {
retryLater(record);
}
注意：即使 send() 返回的 Future 成功，也只代表"Broker 已确认"，不代表下游已消费。业务上需要更强保证时，靠消费端回执或对账，而不是等 get()。
五、常见追问
- 为什么要两条线程？ 主线程不能被网络阻塞，否则业务吞吐被 RTT 限制；Sender 单线程批量发送能把多个分区的请求合并到一次网络往返。
- 重试发生在哪一步？ Sender 内部，根据
retries与delivery.timeout.ms决定是否重发；重试的批次会重新入队，可能与新批次交错，所以需要幂等保证顺序与去重。 - 分区器怎么选分区？ 指定 key 时
murmur2(key) % numPartitions；未指定 key 时 2.4+ 使用粘性分区（Sticky Partitioner），攒满一批再换分区，减少小批次。 send()返回成功但 Broker 没收到，可能吗？ 不可能。回调成功意味着 Broker 已按acks返回确认；如果回调超时，消息可能已写入也可能没写入，只能靠重试 + 幂等。- 拦截器里能做限流吗？ 可以但不推荐，拦截器运行在主线程，任何阻塞都会直接拖慢业务线程。
小结
- 发送是"主线程入缓冲、Sender 线程批量网络 IO"的两段式异步模型。
- 批量大小由
batch.size与linger.ms共同决定，缓冲上限由buffer.memory控制。 - 分区器决定消息落到哪个分区，也就决定了顺序与并行度。
- 回调成功只代表 Broker 确认，端到端可靠仍要靠消费确认与补偿。