Theory · Kafka · Producer Internals
主线程 + Sender 双线程 · RecordAccumulator 攒批 · 幂等去重 · 事务原子 —— 发送侧的全部机制深挖
send() 异步入队：拦截器 → 序列化 → 分区 → 累加器；Sender 单线程按 Node 攒请求发出，回调在 IO 线程触发
acks + 重试 → 幂等（单分区去重，默认开启）→ 事务（跨分区原子 + 跨会话 fencing）
结论对照 Kafka 4.3 文档与 ProducerConfig.java 源码（2026-08），版本差异逐处标注
Send Pipeline
RecordAccumulator · Batching
每个 <topic-partition> 一条 deque，首条消息按 batch.size=16KB 预分配内存。批"封口"后不再接收追加，转为可发送。触发条件：批内字节数 ≥ batch.size、或 linger.ms=5（4.0 起默认，此前 0）到期、或缓冲不足、或 close()。
总缓冲默认 32MB，是主线程写、Sender 读的环形背压。打满时 send() 阻塞等待空闲内存，上限 max.block.ms=60000——超时抛 TimeoutException。首个 send() 拉取元数据的等待也计入 max.block.ms。
自由池按 batch.size 粒度预分配 ByteBuffer：批发出即 deallocate 归还，下个批直接复用——生产者长时间高吞吐运行不产生内存分配 GC 抖动。代价：消息 > batch.size 时绕过池分配整块内存，易碎片化、浪费 buffer.memory。
| 追问 | 答案 |
|---|---|
| batch.size 与 linger.ms 怎么配合？ | 两个独立触发器：空间到量（16KB 满）或时间到点（5ms）先到先封口。吞吐向调法：batch.size 64–512KB + linger.ms 50；延迟向：batch.size 小 + linger.ms 0–5 |
| 压缩发生在哪一层？ | 以整个 ProducerBatch 为单位在发送前压缩（lz4/zstd/gzip/snappy），broker 校验后原样落盘转发，consumer 解压——端到端只压一次（详见 kafka-internals 总览 deck） |
| 单条大消息怎么办？ | 大于 batch.size 的消息独占一个批，内存绕过 free 池按需分配；上限受 max.request.size=1MB 与 broker message.max.bytes 双重约束 |
Sender · InFlightBatches
| 环节 | 机制 |
|---|---|
| ready 判定 | 遍历各分区 deque：有 leader、linger 到期或批已封口、缓冲未满 → 该 node 进入本轮可发集合 |
| drain 按 node 聚合 | 同一 node 的多个分区批合成一个 ProduceRequest（上限受 max.request.size 与 in-flight 约束）——网络放大系数被压到最低 |
| InFlightBatches | 每连接最多 max.in.flight.requests.per.connection=5 个在途请求；超出需等 ack 才 drain 下一轮 |
| 响应处理 | ack 成功 → 完成 batch、触发回调、内存归还池；失败 → 分类后重试或回调异常 |
| 异常分类 | 典型例子与去向 |
|---|---|
| 可重试 RetriableException | NotLeaderForPartitionException、NotEnoughReplicasException、TimeoutException、CoordinatorNotAvailableException…… 按 retries 重试（默认 Integer.MAX_VALUE），受 delivery.timeout.ms 兜底 |
| 不可重试 | RecordTooLargeException、TopicAuthorizationException、InvalidTimestampException…… 立即完成 batch 并回调异常 |
| 幂等相关 | OutOfOrderSequenceException（不可重试，需重建 producer）；DuplicateSequenceException（良性，等同成功） |
delivery.timeout.ms=120000 ≥ linger.ms + request.timeout.ms（文档明确要求）。它覆盖"从 send 成功入队到最终成功或放弃"的全部时间，重试次数不必手调——retries 给到 MAX，由 delivery.timeout 兜底即可。Lifecycle · Timeout Budget
| 阶段 | 发生什么 | 受控参数 | 失败表现 |
|---|---|---|---|
| 1 入队 | 校验 → 序列化 → 分区 → append；缓冲不足则在此阻塞 | max.block.ms=60s | send() 直接抛 TimeoutException |
| 2 攒批 | 在 deque 中等待封口（批满或 linger 到期） | batch.size=16KB · linger.ms=5 | 无（纯延迟项） |
| 3 发送 | drain 聚合 → NIO 写出 → 等待 broker ack | request.timeout.ms=30s | 可重试异常 → 回队重试 |
| 4 重试 | 可重试异常后 batch 重新入队（幂等下 seq 不变） | retries=MAX_VALUE | 受阶段 5 总预算约束 |
| 5 总预算 | 覆盖 2+3+4 的总时间；到期放弃并回调异常 | delivery.timeout.ms=120s | 回调 TimeoutException，消息丢失（若上层不补偿） |
| 6 回调 | onCompletion(metadata, exception) 在 Sender 线程执行 | — | 业务侧唯一可靠的"终态"通知点 |
send() 返回 Future 只代表"已入队"。终态只有两个：回调收到 metadata（成功）或收到异常（放弃）。不检查回调/Future 的生产代码，等于把可靠性降级为 acks=0——这是排查"为什么 Kafka 丢消息"的第一站。
要低延迟：delivery.timeout 适度调小，快速失败快速补偿；要高可靠：delivery.timeout 覆盖"至少 N 轮重试 + 一次 leader 切换"的时长，配合幂等防重复。request.timeout.ms 是单请求项，不要拿它当总预算用。
Partitioning · KIP-794
| 优先级 | 逻辑 | 要点与坑 |
|---|---|---|
| ① 指定分区 | send(record, partition) 显式指定 | 最高优先级，绕过一切策略 |
| ② 按 key 哈希 | toPositive(murmur2(keyBytes)) % numPartitions | murmur2：快、分布均匀；扩分区后同 key 换分区，破坏同 key 保序——规划期定好分区数 |
| ③ 无 key：粘性分区 | 3.3 起内置（partitioner.class 默认 null）：粘住一个分区直到该批攒满 batch.size 字节才切换 | KIP-794：解决老轮转 RoundRobin"每消息换分区 → 批攒不满"的问题，批更大、延迟更低、分布严格均匀 |
| ④ 自定义 | 实现 Partitioner 接口，配置 partitioner.class | 设置后覆盖内置逻辑；DefaultPartitioner / UniformStickyPartitioner 均已 @Deprecated，改为内置实现 |
旧逻辑无 key 消息轮转发：N 个分区每分区摊到一点点消息，批永远攒不满 16KB，只能靠 linger 超时发出小批。粘性分区把"同一时间段的无 key 流量"聚到同一分区，批接近 batch.size 才换——同样的流量，请求数骤降、端到端压缩比也更高（同批字段相似冗余多）。
3.3 起 partitioner.class 默认从 DefaultPartitioner 改为 null：keyed 记录仍 murmur2 哈希，无 key 记录走内置严格均匀粘性分区。补充开关 partitioner.ignore.keys=false（默认）：设 true 时连有 key 的记录也忽略 key 改走粘性均匀（老 UniformStickyPartitioner 的语义，该类已废弃）。
acks · Exceptions
| acks | 客户端语义 | 风险与适用 |
|---|---|---|
| 0 | 网络写出即视为成功，不看任何响应，也基本不重试；回调几乎立即以成功触发 | 日志采集类可容忍丢失场景；吞吐最高、可靠性最弱 |
| 1 | 仅等 leader 副本落盘（内存/页缓存）即 ack；leader 挂且消息未及复制 → 丢 | 折中默认值的历史选项；3.0 起 acks 默认已改 all |
| all(-1) | 等 ISR 全部副本写入才 ack；配合 broker 端 min.insync.replicas：ISR 数量不足时直接拒写（NotEnoughReplicas） | 3.0+ 默认；"all 只等 ISR 不等 AR"的细节见 replication deck |
onCompletion(exception, metadata) 里 exception 非空即终态失败。工程上三选一：重试类交给 producer 自动重试（幂等下不重复）；不可重试类（RecordTooLarge、鉴权失败）改数据或告警；超时类落本地补偿表异步重投。任何路径都要有——静默丢回调是最常见的事故源。
enable.idempotence=true（3.0+ 默认）时：acks 默认 all、retries 默认 MAX、max.in.flight ≤ 5。显式把 acks 改为 0/1 而不显式关闭幂等 → ConfigException 启动失败；显式 enable.idempotence=false 才能自由降级 acks——版本升级踩坑高发点。
Idempotence · PID + Sequence
// RecordBatch 头（magic=2）中幂等三字段 producerId: int64 // PID：InitProducerId 分配 producerEpoch: int16 // 会话代次，fencing 用 baseSequence: int32 // 批首条序列号 // broker 端 ProducerStateManager： // 每 <PID, partition> 保留最近 5 个批的 // 元数据（ProducerStateEntry#NUM_BATCHES_TO_RETAIN=5） if baseSeq == lastSeq + 1 // 连续 → 接受 else if baseSeq ≤ lastSeq // 区间命中 → Duplicate // 静默丢弃，返回成功 else // 出现缺口 → 抛 // OutOfOrderSequenceException
seq 从 0 起按分区递增；重试批的 seq 与首战相同，broker 由此识别重复
| 要点 | 说明 |
|---|---|
| PID 怎么来 | producer 初始化时向 broker 发 InitProducerId 获取；无事务也隐式获取。重启进程 → 新 PID（老 PID 作废） |
| 解决什么 | 重试重复：request 超时但实际已写入 → 重试批被 broker 识别为 Duplicate 丢弃并返回成功——"半成功"不再放大成重复 |
| 默认开启 | enable.idempotence=true（3.0+，KIP-679）；联动约束：acks=all + retries>0 + max.in.flight ≤ 5 |
| 去重范围 | broker 端缓存窗口内（最近 5 个批）；窗口外/日志已清理的重复不在覆盖范围（极老批的重复由日志层去重兜底） |
Idempotence · Boundaries
序列号按 <PID, partition> 维护，去重与保序都只在分区内成立。跨分区没有任何顺序承诺——业务有序必须按 key 路由到同一分区。
producer 重启换新 PID，broker 无从关联旧会话——跨会话的重复（如旧进程残留请求）与乱序不被幂等覆盖。需要跨会话强保证时上事务：transactional.id + epoch fencing 旧会话。
幂等只到 broker 落盘为止。消费者重消费、下游写库等"业务重复"仍需消费端位移管理或下游幂等（唯一键/版本号）兜底——端到端 exactly-once 是事务 + read_committed 的事。
开启幂等后，批允许乱序到达，但 broker 缓存了该 PID 每分区最近 5 个批的 seq 区间：乱序批按序列号校验连续性、重复批被丢弃，落盘顺序由 seq 决定——客户端无需退化为串行。5 不是拍脑袋：正好等于 broker 保留的批元数据数量。
在途批超过 5，缺口可能落在 broker 缓存窗口之外：baseSeq 与缓存末序列号对不上、无法判定"重发还是乱序"——幂等前提失效。配置校验直接拦截：幂等开启且 max.in.flight > 5 → ConfigException。未开启幂等时 >5 不报错，但乱序真乱序。
Transactions · KIP-98
Semantics · LSO · read_committed
| 保证 | 机制 |
|---|---|
| 跨分区/跨会话原子写 | 一批跨多分区的消息要么全部对 read_committed 消费者可见，要么全部不可见——COMMIT/ABORT 标记（控制批）落在各分区日志 |
| 僵尸 producer 隔离 | transactional.id 复用 → InitProducerId 将 epoch+1；旧 epoch 实例的写入被 broker 以 InvalidProducerEpoch/ProducerFencedException 拒绝（幂等单会话做不到这层） |
| 消费隔离 read_committed | isolation.level=read_committed 时只读 LSO 之前且未被 abort 事务覆盖的记录；LSO = 最早未决事务的起点，挂起的事务会卡住 LSO、抬高消费 lag |
| 消费位移同事务绑定 | sendOffsetsToTransaction 把"消费位移提交"并入本事务 → 消费-转换-生产链路端到端 exactly-once（Kafka Streams EOS 的基础） |
事务包含幂等：设了 transactional.id 即强制 enable.idempotence。幂等解决"单分区、单会话"的重试去重；事务在此之上加三件事——跨分区原子、跨会话 fencing（epoch）、位移绑定。答"事务是不是为了不丢"就错了，它解决的是原子性与隔离性，不丢靠 acks+ISR。
transaction.timeout.ms=60000（默认，broker 上限 transaction.max.timeout.ms=900000）；事务状态写 __transaction_state（RF=3、min.isr=2）→ 要求集群至少 3 broker；每事务多轮 RPC + 标记批，吞吐低于裸幂等。事务粒度别太细：一批一次 commit 是常见反模式。
Cheatsheet · Defaults（4.3 源码核实）
| 可靠性 / 行为 | 默认 | 调优建议 |
|---|---|---|
| acks | all | 金融链路保持 all + min.insync.replicas=2（topic 级）；日志类可降 1 |
| enable.idempotence | true | 保持默认；显式降 acks 时需显式关幂等 |
| retries | MAX_VALUE | 保持默认，用 delivery.timeout 兜底 |
| delivery.timeout.ms | 120000 | ≥ linger + request.timeout；覆盖至少一轮 leader 切换 |
| request.timeout.ms | 30000 | 跨机房可适度调大，非总预算 |
| max.in.flight.requests.per.connection | 5 | 幂等下 ≤5；非幂等降为 1 可严格串行（牺牲吞吐） |
| 性能 / 事务 | 默认 | 调优建议 |
|---|---|---|
| batch.size | 16384 | 吞吐向 64–512KB；注意大消息绕池的碎片问题 |
| linger.ms | 5（4.0 起，此前 0） | 吞吐向 50；低延迟链路 0–5 |
| buffer.memory | 33554432 | 大批量/多分区场景加大，防 max.block 超时 |
| max.block.ms | 60000 | 按"可接受的 send 阻塞上限"设置 |
| compression.type | none | lz4 均衡推荐；zstd 压缩比最高（2.1+） |
| max.request.size | 1048576 | 与 broker message.max.bytes、topic max.message.bytes 三者对齐 |
| transactional.id | null | 设置即强制幂等；transaction.timeout.ms 默认 60000（broker 上限 900000） |
Interview QA · 1/2
线程安全：单实例可多线程并发 send。主线程做拦截器→序列化→分区→append 入累加器；Sender 单守护线程负责 drain、网络收发；回调也在 Sender 线程触发，回调里勿做重活。
两个独立封口触发器先到先得：批内字节 ≥ batch.size（16KB）或 linger.ms（4.0 起默认 5ms）到期。吞吐向调 64–512KB + 50ms；延迟向调小。压缩比随批变大而提升。
send() 阻塞等待空闲内存，上限 max.block.ms=60s，超时抛 TimeoutException（消息未入队、需业务重试）。首次 send 等元数据的时间也计入。缓解：加大 buffer.memory、提高消费速度（Sender）、压批。
不开启幂等会：超时但实际已写入 → 重试落两条。幂等（默认开启）下重试批 seq 不变，broker 端识别 Duplicate 静默丢弃并返回成功——重试不重复、不乱序。
不一定。all 只等当前 ISR：ISR 收缩到 1 时退化为 acks=1。需 broker 端 min.insync.replicas ≥2 拒绝低冗余写入，且不开启 unclean 选举；客户端还要处理回调异常——三件套缺一不可（矩阵见 replication deck）。
每批带 PID + epoch + baseSequence；broker 为每 <PID,partition> 缓存最近 5 个批的 seq 区间：连续接受、重复丢弃返回成功、缺口抛 OutOfOrderSequence。解决"重试重复"。
乱序到达时 broker 按 seq 区间校验连续性、丢弃重复，落盘顺序由 seq 决定，客户端无需串行。5 正好等于 broker 每分区每 PID 保留的批元数据数；>5 缺口可能落在窗口外，无法判定重发或乱序 → 配置校验直接报错。
①序号按分区维护，只保分区内有序；②重启换 PID，跨会话重复/乱序不覆盖（需事务 fencing）；③只到 broker 落盘，消费侧重复要靠位移管理或下游幂等。业务跨分区有序必须按 key 路由同分区。
Interview QA · 2/2
init：FindCoordinator → InitProducerId（epoch+1，fence 旧会话）。事务体：AddPartitionsToTxn 登记 + 带事务头的批。commit 两阶段：先 PREPARE_COMMIT 持久化到 __transaction_state（裁决点），再向各分区写 COMMIT 控制批，落 COMPLETE_COMMIT 后回执。
同一 transactional.id 再初始化时 epoch+1；旧 epoch 实例的写入被 broker 拒绝（ProducerFenced/InvalidProducerEpoch）。分区 leader 也会拒绝低于当前记录 epoch 的批——挂在 GC 停顿里的旧实例无法继续污染日志。
只读 LSO 之前、未被 abort 事务覆盖的记录。LSO 是最早未决事务的起点：事务长时间不 commit/abort 会卡住 LSO，read_committed 消费者 lag 持续上涨——排查 lag 先看有无挂起事务。
事务包含幂等（设 transactional.id 强制幂等）。幂等：单分区单会话去重；事务额外提供跨分区原子、跨会话 fencing、消费位移绑定。事务解决原子性/隔离性，"不丢"仍靠 acks+min.insync+ISR。
toPositive(murmur2(keyBytes)) % numPartitions：murmur2 快且均匀。分区数改变后同 key 落新分区：既有顺序保证破坏、局部性缓存失效——需要按 key 保序的主题应在规划期定好分区数（或用足够大的初始分区数）。
老轮转发每消息换分区，批攒不满只能靠超时发小批。粘性分区粘住一个分区直到攒满 batch.size 再换：请求更少、批更大、压缩比更高、分布仍严格均匀。3.3 起 partitioner.class 默认 null，DefaultPartitioner 废弃。
拦截器适合横切关注点：打点监控（发送数/耗时）、统一加 header、脱敏清洗；onAcknowledgement 做成功率统计。注意 onSend 可改写记录影响下游。序列化器做对象↔字节，推荐 Avro/Protobuf+Schema Registry，避免 JSON 体积与演进问题。
用途：Kafka Streams 端到端 exactly-once（消费位移并入事务）、跨多分区原子写、CDC 一致性投递。代价：每事务多轮 RPC + 标记批、__transaction_state 需 RF=3/min.isr=2（至少 3 broker）、事务超时上限 15 分钟；勿一消息一事务。
Related & References
send() → 攒批（batch.size/linger.ms）→ Sender drain
分区：key murmur2 / 无 key 粘性（KIP-794）
可靠性：acks=all + min.insync → 幂等去重 → 事务原子
重复与乱序 → PID+seq broker 校验（in-flight ≤5）
端到端 exactly-once → 事务 + read_committed + 位移绑定
参考来源（版本敏感结论均可溯源，2026-08 核实）
| kafka.apache.org/documentation/#producerconfigs | Producer configs：acks/retries/delivery.timeout/max.block.ms 等默认值与约束说明 |
| github.com/apache/kafka（4.3）ProducerConfig.java | 默认值源码级核实：linger.ms=5（4.0 由 0 改 5）、enable.idempotence=true（幂等三约束）、max.in.flight=5、partitioner.class=null |
| cwiki.apache.org/confluence/display/KAFKA/KIP-794 | Uniform Sticky Partitioner：3.3 起内置粘性分区、partitioner.class 默认 null、旧 Partitioner 类废弃 |
| cwiki.apache.org/confluence/display/KAFKA/KIP-679 | Producer will enable idempotence by default：3.0 起默认开启及 acks/retries/max.in.flight 联动约束 |
| cwiki.apache.org/confluence/display/KAFKA/KIP-98（Exactly Once Delivery and Transactional Messaging） | 事务协议：InitProducerId/AddPartitionsToTxn/EndTxn 两阶段与控制批 |
| kafka.apache.org/documentation/#semantics · #transactional_format | 交付语义、read_committed 与控制批/标记格式 |