4. Kafka 的 Exactly-Once 语义：事务性生产者、幂等生产者、事务协调器的工作原理，以及如何配置才能保证不丢不重

聊到实时数仓的数据一致性，Kafka 的 Exactly-Once 语义是个绕不开的话题。很多朋友问我：“Kafka 不是号称高吞吐吗？怎么保证数据不丢不重？”

其实，Kafka 的 Exactly-Once 并不是一个单一的特性，而是一套组合拳。它由三个核心组件构成：幂等生产者、事务性生产者和事务协调器。我最早接触这块时，也踩过不少坑，今天就把这些经验掰开揉碎讲清楚。

4.1 幂等生产者：最基础的“去重”保障

幂等生产者，说白了就是让 Kafka 能识别并丢弃重复的消息。它的原理其实很简单：每个生产者启动时，会被分配一个唯一的 ProducerId（PID）。每次发送消息时，消息里会带上一个单调递增的 Sequence Number。

Broker 端会为每个 PID 维护一个内存中的序列号状态。当收到一条消息时，Broker 会检查它的序列号：

• 如果序列号比当前记录的大 1，说明是正常的新消息，接受并更新状态。
• 如果序列号比当前记录的小，说明是重复消息，直接丢弃。
• 如果序列号比当前记录的大很多，说明有消息丢失，Broker 会报错。

# 生产者端配置
enable.idempotence = true
# 以下配置在开启幂等后会自动优化，但建议显式设置
acks = all
max.in.flight.requests.per.connection = 5  # 幂等模式下支持乱序，但建议保持默认
retries = Integer.MAX_VALUE

我记得有一次，一个同事在生产环境只开了幂等，没开事务，结果在重启生产者后出现了少量重复。为什么？因为生产者重启后 PID 会变，Broker 端无法跨 PID 去重。所以，幂等生产者只能保证单次会话内的 Exactly-Once，跨会话就不行了。

4.2 事务性生产者：跨分区、跨会话的原子性保证

事务性生产者，才是真正实现端到端 Exactly-Once 的关键。它解决了幂等生产者无法处理的场景：比如一个事务要写入多个分区，或者生产者崩溃后重启，需要保证之前未提交的事务要么全部成功，要么全部回滚。

它的核心思想是：引入一个事务协调器（Transaction Coordinator），负责管理事务的生命周期。

4.2.1 事务协调器的工作原理

事务协调器是运行在 Broker 上的一个模块。每个生产者通过 transactional.id 与协调器绑定。协调器会做以下几件事：

1. 分配 ProducerId：生产者首次初始化时，协调器分配一个 PID，并将 transactional.id 与 PID 的映射关系写入内部主题 __transaction_state。
2. 管理事务状态：协调器维护每个事务的状态（如 Ongoing、PrepareCommit、CompletedCommit 等），并持久化到 __transaction_state 主题。
3. 处理事务超时：如果生产者崩溃，协调器会检测到事务超时，并主动回滚未完成的事务。
4. 协调提交/回滚：生产者调用 commitTransaction() 时，协调器会先写入一个 PrepareCommit 标记，然后等待所有参与分区的 Leader 确认，最后写入 CompletedCommit 标记。

4.2.2 事务性生产者的配置

# 生产者端配置
enable.idempotence = true          # 事务必须开启幂等
transactional.id = "my-transactional-id"  # 唯一标识，用于恢复会话
acks = all
# 消费者端配置（用于事务性消费）
isolation.level = read_committed   # 只读取已提交的事务消息

配置好之后，代码里需要这样用：

// 初始化事务生产者
producer.initTransactions();

try {
    // 开始事务
    producer.beginTransaction();
    
    // 发送消息到多个分区
    producer.send(new ProducerRecord<>("topic-a", "key1", "value1"));
    producer.send(new ProducerRecord<>("topic-b", "key2", "value2"));
    
    // 提交事务
    producer.commitTransaction();
} catch (Exception e) {
    // 回滚事务
    producer.abortTransaction();
}

4.3 事务性消费者：如何做到“不丢不重”

光有生产者端的事务还不够，消费者端也得配合。Kafka 提供了 read_committed 隔离级别，消费者只会读取已提交的事务消息，未提交或已回滚的消息会被过滤掉。

但这里有个细节：消费者在消费事务消息时，如果消费进度（Offset）提交到了非事务消息中，一旦消费者崩溃重启，可能会重复消费已提交的事务消息。所以，我建议将消费进度也纳入事务管理。

// 在事务内提交消费进度
producer.beginTransaction();
// 消费消息
ConsumerRecords records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord record : records) {
    // 处理业务逻辑
    process(record);
    // 将消费进度作为消息发送到内部主题
    producer.send(new ProducerRecord<>("__consumer_offsets", 
        record.partition(), record.offset()));
}
// 提交事务，同时提交消费进度
producer.commitTransaction();

你想想看，这样做的好处是：如果消费者在处理过程中崩溃，事务会回滚，消费进度也不会提交。下次重启时，消费者会重新消费未提交的消息，保证了 Exactly-Once。

4.4 核心流程图：事务性生产者的完整生命周期

下面这张图，是我根据实际项目经验画的。它展示了事务从初始化到提交/回滚的完整流程：

4.5 配置清单：如何做到“不丢不重”

说了这么多，到底怎么配？我整理了一份清单，照着做基本不会出问题：

4.6 避坑指南

最后，分享几个我实际项目中遇到的坑：

• 事务超时导致数据丢失：默认事务超时是 1 分钟。如果业务处理时间超过 1 分钟，事务会被协调器强制回滚。我建议根据业务情况调整 transaction.timeout.ms。
• 消费者 Offset 提交不一致：如果消费者在事务外提交 Offset，一旦消费者崩溃，重启后会重复消费。一定要把 Offset 提交也纳入事务管理。
• 跨集群复制时的陷阱：MirrorMaker 2 不支持事务消息的跨集群复制。如果你需要跨集群同步，得自己处理事务状态。

嗯，Kafka 的 Exactly-Once 语义，说白了就是一套“去重 + 原子提交”的机制。幂等生产者解决单次会话的重复，事务性生产者解决跨会话的原子性，事务协调器负责兜底。配置对了，数据不丢不重是完全可以做到的。