Producer 与投递语义

一句话：Producer 决定一条消息「什么时候算发送成功」，以及重试、重启、跨分区写入时会不会重复或只成功一半。

适用版本：Kafka 3.9 / 4.0（Java 客户端）。默认值以 3.0+ 为准。

为什么需要

网络会超时，Broker 会切换，Producer 会重启。每次超时都重发，消息可能重复；不重发，消息可能丢。只靠 send() 的返回值无法回答「到底写进去没有」。

投递语义把这几个问题拆开：acks 决定写到什么程度算成功；幂等决定同一次发送的重试会不会重复；事务决定多条消息、多个分区能否一起可见或一起回滚。

核心机制

发送路径

send() 先进本地缓冲区（buffer.memory），按分区攒成批次（batch.size），由 Sender 线程发出；linger.ms 决定没攒满时最多等多久。批次越大、等待越久，吞吐越高、单条延迟越大；缓冲区满时 send() 会阻塞 max.block.ms（默认 60000）。

acks 三档

acks	成功条件	风险
`0`	发出即认为成功	Broker 没收到也不知道，可能丢
`1`	Leader 写入本地日志	Leader 挂、Follower 未同步时丢
`all` / `-1`	ISR 内所有副本都写入	延迟最高；ISR 缩小时仍可能退化

3.0 起 acks 默认是 all（KIP-679），不再是 1。

acks=all 单独用不够：如果 ISR 只剩 Leader，它等于 acks=1。还要配 min.insync.replicas=2，ISR 不足时 Producer 会收到 NotEnoughReplicasException；再配 unclean.leader.election.enable=false，避免非 ISR 副本当选 Leader。三者合起来才是「不丢已提交消息」，代价是挂两台时该分区拒写。

重试与 delivery.timeout.ms 的预算

retries：默认 2147483647（Integer.MAX_VALUE），3.0 起配合幂等默认开启。
delivery.timeout.ms：默认 120000，从 send() 到收到最终结果的总预算，包含攒批、重试、在途请求的时间。
request.timeout.ms：默认 30000，单次请求的超时；retry.backoff.ms 默认 100。

delivery.timeout.ms 是总闸，retries 只是次数上限。超过总预算，消息以失败告终，回调收到 TimeoutException；此时不要盲目再发，先确认是否已经写入（靠业务幂等键或事务）。配置校验要求 delivery.timeout.ms >= linger.ms + request.timeout.ms。

幂等：PID + sequence

开启 enable.idempotence=true（3.0 起默认）后：Producer 从 Broker 申请一个 PID（Producer ID）和 epoch；每个 <PID, 分区> 维护一个单调递增的 sequence；Broker 按 <PID, 分区, sequence> 去重，重复的 sequence 直接返回成功，不重复写。

所以网络超时后的重试不会在 Broker 端产生两条。前提是 acks=all、retries > 0、max.in.flight.requests.per.connection <= 5，否则客户端直接报配置错误。

在途请求不超过 5 时，幂等保证重试后分区内仍然有序。不需要把 max.in.flight 改成 1，那是没有幂等时的旧做法。

边界：幂等只覆盖单个分区、单次 Producer 会话。跨分区不保证原子；Producer 重启后 PID 会变，新会话的 sequence 从 0 开始，旧会话的重试无法去重。

事务：跨分区原子写入

事务把「多条消息 + 消费位点」放进同一个原子单元。Producer 配置 transactional.id，这个 ID 是跨重启的稳定身份，Broker 用它做 fencing：同一 ID 的新实例初始化时 epoch 加一，旧实例再写入会收到 ProducerFencedException。流程是 initTransactions() 注册并获取 PID/epoch → beginTransaction() → 多次 send()（可跨分区）→ commitTransaction()；任一步失败调 abortTransaction()，未提交的消息对 read_committed 消费者不可见。消费-处理-生产模式还要用 sendOffsetsToTransaction() 把消费位点纳入同一个事务。

可见性由 LSO（Last Stable Offset）控制：read_committed 消费者只能读到 LSO 之前的消息，LSO 是最早一个未完成事务的起始 offset。事务开得越久，LSO 越靠前，下游越滞后。默认 Producer transaction.timeout.ms=60000，Broker transaction.max.timeout.ms=900000；超过超时，协调器会中止事务。

事务不等于业务事务：它只保证 Kafka 的写入和位点原子，不覆盖数据库、HTTP 调用、缓存。外部副作用仍需业务幂等（唯一键、去重表、outbox）。

三种投递语义

acks=0 或先提交后处理是最多一次；acks=all + 重试 + 先处理后提交是至少一次；幂等 + 事务 + read_committed + 位点入事务才是 Kafka 内的精确一次，仅限 Kafka 内部链路。

实战场景

一、`send()` 把 Web 线程阻塞 60 秒

现象：大促开始后交易接口 P99 从 60ms 涨到 60 秒以上，Tomcat 线程全部繁忙，连不依赖 Kafka 的查询接口也超时。Broker 侧监控正常、写入速率没有明显下降；应用日志里出现 TimeoutException: Expiring 1 record(s) ... 120000 ms has passed since batch creation。

排查：send() 不是“立即发出”，它先写本地缓冲区。

定位层	取证据	判定
线程栈	业务线程是否停在 `KafkaProducer.send` → `RecordAccumulator.append` → `waitOnBufferAvailable`	卡在这里就是缓冲区满
缓冲区水位	客户端指标 `buffer-available-bytes`、`waiting-threads`	available 长期为 0 即打满
单批发出速度	`batch-size-avg`、`record-send-rate`、`request-latency-avg`	请求延迟高 → 攒批发不出去
`max.block.ms`	默认 60000	阻塞时长与 P99 吻合即确认
`delivery.timeout.ms`	默认 120000，日志里的 `120000 ms has passed`	批次在缓冲区里就过期，说明发送速度远低于生产速度
Broker/网络	ISR、Leader 切换、跨机房带宽	排除下游变慢这个诱因

根因链条通常是：Broker 端某个分区 Leader 切换或磁盘变慢 → Sender 线程发不出去 → 32 MiB 的 buffer.memory 填满 → 业务线程在 send() 上等到 max.block.ms，把 Web 线程池一起拖死。

对症做法：

java

// 1. 不要在请求线程里等 Kafka：异步发送 + 回调，主流程只负责落库
producer.send(record, (meta, ex) -> {
    if (ex != null) {
        metrics.sendFailed(ex.getClass().getSimpleName());
        outbox.markFailed(meta == null ? record.topic() : "", ex);   // 交给补偿，不阻塞业务
    }
});
// 同步等待结果只用在真正需要确认的场景（对账、退款），并接受它占线程

// 2. 把“阻塞 60 秒”改成“快速失败 + 降级”
p.put("max.block.ms", "2000");            // 最多等 2s 拿缓冲区，超了就抛
p.put("delivery.timeout.ms", "15000");
p.put("buffer.memory", "67108864");        // 64 MiB，配合下面三项

// 3. 提升单批效率而不是先加线程
p.put("linger.ms", "20");
p.put("batch.size", "65536");
p.put("compression.type", "lz4");

text

4. 业务侧要有降级路径：事件写入失败时落本地消息表，由后台任务重投；
   Kafka 的可用性不该等于交易接口的可用性。
5. 若必须同步发送，给 send 单独设线程隔离（独立线程池 + 超时），
   别让一个 Topic 的抖动带走整个进程。

复盘要点：判据是“业务线程池里不允许出现停在 waitOnBufferAvailable 的线程”，可以直接从线程栈采样验证。把 buffer-available-bytes、waiting-threads、record-error-rate 三个客户端指标接进告警——它们在用户可感知的超时之前就会变红。max.block.ms 默认 60 秒是入门值不是生产值，这一条要写进客户端封装的默认配置。

二、幂等开着，重复扣款还是发生了

现象：风控回调接口对同一笔订单发了两次 send()，下游账务收到两条内容相同、offset 相邻的消息，触发两次扣款。团队最初的结论是“enable.idempotence 不是保证不重复吗”，于是把幂等关掉又打开，问题仍在。

排查：幂等去重的键是 <PID, 分区, sequence>，它只认“同一次发送的重试”。

定位层	取证据	判定
两次写入的 sequence	`kafka-dump-log.sh --print-data-log` 看两条记录的 producerId/sequence	sequence 不同 → 是两次独立发送，不是重试
调用次数	应用日志/trace 里同一业务请求调用了几次 `send()`	两次调用即为应用层重复
触发来源	上游 HTTP 重试、消息重投、用户重复提交、事务重开	都在幂等范围之外
Producer 会话	两次发送是否跨进程重启	重启后 PID 变化，sequence 从 0 开始，无法去重
下游是否有唯一键	账务表是否有业务幂等键与去重逻辑	无键则重复必然生效

幂等覆盖的是“网络超时后客户端自己重发同一条”，不覆盖“业务把同一条消息发了两遍”，也不覆盖跨分区与跨重启。

对症做法：

java

// 1. 业务幂等键必须进消息，而不是靠 Kafka 去重
public class AccountCommand {
    @NotBlank String commandId;     // 由上游生成并在重试间复用，不是每次新生成
    long orderId; BigDecimal amount;
}

// 2. 下游用唯一索引收敛“至少一次”
// INSERT ... ON CONFLICT (command_id) DO NOTHING
boolean first = ledger.insertIfAbsent(cmd.commandId(), cmd);
if (!first) return Ack.duplicate();          // 重复直接确认，不再执行副作用

// 3. 上游重试时复用同一个 commandId：这是“幂等”能成立的前提
String commandId = request.idempotencyKeyOrGenerate();   // 从请求头取，取不到才生成

text

4. 需要跨分区原子（订单 + 库存一起可见）时用事务，不靠幂等；
   事务也只管 Kafka，不包数据库写入，见下节 LSO 的代价。
5. 消息体里带 schema 版本与 commandId，便于事后判断“重复”从哪来。

复盘要点：语义要说准——acks=all + 重试是至少一次，幂等 Producer 去掉的是重试造成的分区内重复，read_committed + 事务把范围收紧到 Kafka 内部的精确一次。只要链路上有数据库或外部调用，正确性就落在业务幂等键上。判据是“同一 commandId 在下游只生效一次”，用一条重复投递的测试固化，见可靠事件投递：本地消息表与幂等。

三、一个长事务把下游延迟拖到 15 分钟

现象：实时看板（isolation.level=read_committed）的数据比订单库晚 15 分钟，而消费者 lag 显示正常。同时 Kafka 侧 AbortablePrepare 相关日志增多。根源是新建的“批量结算 Producer”在事务里做了一次外部 HTTP 调用，单事务时长最长达到 8 分钟。

排查：LSO 是最早一个未完成事务的起始 offset，一个长事务会挡住其后所有已提交消息的可见性。

定位层	取证据	判定
消费者可见延迟 vs lag	lag 很小但读取位点长期不动	典型的 LSO 被卡，不是消费慢
未完成事务	`kafka-transactions.sh --describe --transactional-id set-batch-1`	状态 `Ongoing` 且持续时间长即命中
事务内动作	代码里 `beginTransaction` 与 `commit` 之间是否有 HTTP/DB/循环等待	有外部调用就是设计问题
超时配置	Producer `transaction.timeout.ms`（默认 60000）与 Broker `transaction.max.timeout.ms`（默认 900000）	事务超过 Producer 默认值会被协调器中止
分区覆盖	一次事务写入的分区数量	卡住的 LSO 是每个分区独立的，但同分区所有后续消息都被挡
fencing	是否多个实例共用同一 `transactional.id`	互相 fence 会留下未提交事务等待中止

对症做法：

java

// 1. 事务只包 Kafka 写入，外部调用移出事务边界
OrderSnapshot snap = loadSnapshot(orderId);      // HTTP/DB 在事务外完成
producer.beginTransaction();
producer.send(orderRecord(snap));
producer.send(summaryRecord(snap));
producer.commitTransaction();                     // 事务时长 = 两次 send 的时间

// 2. 批量场景拆小事务，而不是一整个批次一个事务
if (++countInTx >= 500) { producer.commitTransaction(); producer.beginTransaction(); countInTx = 0; }

// 3. 超时显式设置并留出余量：Producer 值必须小于 Broker 上限
p.put("transaction.timeout.ms", "30000");     // 单事务上限 30s，超时即中止而非长挂

bash

# 4. 卡住时先看事务状态，再决定 abort；不要直接重启 Producer
bin/kafka-transactions.sh --bootstrap-server broker:9092 \
  --describe --transactional-id set-batch-1
bin/kafka-transactions.sh --bootstrap-server broker:9092 \
  --abort --transactional-id set-batch-1 --transactional-id-timeout-ms 10000

同一 transactional.id 只能有一个活跃实例，分区级并行要用不同 ID；共享受影响的分区上，后续已提交消息的可见性由 LSO 决定，所以“一个慢事务”会连累整个分区的下游。

复盘要点：判据是“事务时长 P99 必须小于 transaction.timeout.ms 的一半”，以及“事务内不存在外部 IO”——后者可以用代码扫描（beginTransaction 与 commit 之间不允许出现 HTTP/DAO 调用）挡下来。给消费端加一个 read_committed 位点与 HW 的差值指标，这个差值就是 LSO 造成的可见性延迟，比 lag 更早暴露问题。

最小上手

Java 客户端（kafka-clients 3.9.x / 4.0.x）的最小生产者：

java

Properties p = new Properties();
p.put("bootstrap.servers", "localhost:9092");
p.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
p.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
p.put("acks", "all");                    // 3.0 起已是默认，显式写出便于审阅
p.put("enable.idempotence", "true");     // 3.0 起已是默认
p.put("linger.ms", "5");
p.put("batch.size", "32768");
p.put("compression.type", "lz4");

try (Producer<String, String> producer = new KafkaProducer<>(p)) {
    producer.send(new ProducerRecord<>("orders", "o-1", "created"), (meta, ex) -> {
        if (ex != null) ex.printStackTrace();
        else System.out.printf("partition=%d offset=%d%n", meta.partition(), meta.offset());
    });
    producer.flush();
}

用 console 工具验证（假设 orders 已创建）：

bash

bin/kafka-console-producer.sh --bootstrap-server localhost:9092 \
  --topic orders --producer-property acks=all
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \
  --topic orders --from-beginning --max-messages 1

事务骨架：

java

p.put("transactional.id", "order-tx-1");
try (Producer<String, String> producer = new KafkaProducer<>(p)) {
    producer.initTransactions();
    try {
        producer.beginTransaction();
        producer.send(new ProducerRecord<>("orders", "o-1", "created"));
        producer.send(new ProducerRecord<>("inventory", "o-1", "reserved"));
        producer.commitTransaction();
    } catch (Exception e) {
        producer.abortTransaction();
        throw e;
    }
}

消费者侧要配 isolation.level=read_committed 才能过滤未提交事务；默认 read_uncommitted 会读到已写入但未提交的消息（这些消息最终可能被 abort）。

生产动作

默认值（3.9 / 4.0 Java 客户端）：

配置	默认值	备注
`acks`	`all`	3.0 起，KIP-679
`enable.idempotence`	`true`	3.0 起
`retries`	`2147483647`	配合幂等
`delivery.timeout.ms`	`120000`	总预算
`batch.size`	`16384`	按分区
`buffer.memory`	`33554432`	全局缓冲
`max.request.size`	`1048576`	单请求上限
`compression.type`	`none`	按带宽/CPU 选 lz4/zstd
`max.in.flight.requests.per.connection`	`5`	幂等下 ≤5 即保序

配置检查顺序：

先确认 Topic 的 min.insync.replicas 和 unclean.leader.election.enable，否则 acks=all 也会退化。
大消息场景同时看 max.request.size、Broker message.max.bytes、Follower replica.fetch.max.bytes；三处任何一处小，都会表现为发送失败或副本追不上。
吞吐不够先调 linger.ms（5–20ms）和 batch.size，再看压缩，不要先加线程。
事务链路保证事务短：事务越长，LSO 越靠后，read_committed 消费者延迟越大；同时确认 transaction.timeout.ms 小于 Broker 的 transaction.max.timeout.ms。
同一 transactional.id 只能有一个活跃实例；分区级并行用不同 ID，避免互相 fencing。

边界

幂等不跨分区、不跨重启。它只保证单分区内重试不重复；跨分区的原子性靠事务，跨重启的身份靠 transactional.id（PID 会变）。
幂等不解决业务重复。应用层重复调用 send()、上游重复投递、消费者重复处理，都不在幂等范围内。
事务不覆盖外部系统。数据库写入、HTTP 调用不会跟着 abortTransaction() 回滚，必须自己做幂等或 outbox。
read_committed 有延迟代价。长事务会卡住 LSO，让下游看不到后续已提交消息。
acks=all 不等于绝对不丢。还要看 ISR 是否真的有两个以上副本、是否允许 unclean 选举，见 01-架构与ISR。

讲清楚

Producer 的投递语义可以按三层说。第一层是 acks：0 是发出去就算，1 是 Leader 写完就算，all 是 ISR 都写完才算。3.0 起默认就是 all，但 all 只有在 min.insync.replicas>=2 且不允许 unclean 选举时才真正防丢，否则 ISR 缩到 1 就退化成 1。第二层是幂等。超时重试会让 Broker 收到重复消息，所以 Producer 申请 PID，每个分区带递增 sequence，Broker 按 <PID, 分区, sequence> 去重。默认开启，max.in.flight<=5 就能在重试时保序，不需要改成 1。它只覆盖单分区、单次会话，重启后 PID 变了就不认旧账。
第三层是事务。transactional.id 给 Producer 一个跨重启的稳定身份，新实例 epoch 加一会 fence 掉旧实例；多条消息和消费位点可以一起提交或回滚，read_committed 消费者只看到 LSO 之前的消息。但事务只包住 Kafka，不包数据库和外部调用。排障时先看 delivery.timeout.ms 这个总预算，再看 acks 和 ISR，最后才是批次和压缩。

Kafka 索引

RocketMQ

ElasticJob

ZooKeeper

XXL-JOB

Producer 与投递语义

为什么需要

核心机制

发送路径

acks 三档

重试与 delivery.timeout.ms 的预算

幂等：PID + sequence

事务：跨分区原子写入

三种投递语义

实战场景

一、`send()` 把 Web 线程阻塞 60 秒

二、幂等开着，重复扣款还是发生了

三、一个长事务把下游延迟拖到 15 分钟

最小上手

生产动作

边界

讲清楚

Producer 与投递语义 ​

为什么需要 ​

核心机制 ​

发送路径 ​

acks 三档 ​

重试与 delivery.timeout.ms 的预算 ​

幂等：PID + sequence ​

事务：跨分区原子写入 ​

三种投递语义 ​

实战场景 ​

一、send() 把 Web 线程阻塞 60 秒 ​

二、幂等开着，重复扣款还是发生了 ​

三、一个长事务把下游延迟拖到 15 分钟 ​

最小上手 ​

生产动作 ​

边界 ​

讲清楚 ​

Producer 与投递语义

为什么需要

核心机制

发送路径

acks 三档

重试与 delivery.timeout.ms 的预算

幂等：PID + sequence

事务：跨分区原子写入

三种投递语义

实战场景

一、`send()` 把 Web 线程阻塞 60 秒

二、幂等开着，重复扣款还是发生了

三、一个长事务把下游延迟拖到 15 分钟

最小上手

生产动作

边界

讲清楚