📚 课程目录第4章 Kafka事务消息

4、Kafka事务消息：Kafka事务API、事务协调器、事务日志、幂等性保证

Kafka 的事务消息，说实话，是很多人在流式处理中绕不过去的一道坎。我刚开始接触时也觉得这东西玄乎，不就是发个消息吗，怎么还扯上事务了？后来在做一个实时对账系统时，才真正体会到它的价值——没有事务，数据对不上，那真是欲哭无泪。

这一章，我们就来拆解 Kafka 事务的四个核心模块：事务 API、事务协调器、事务日志，以及幂等性保证。我会结合自己的踩坑经历，帮你把这些概念串起来。

4.1 事务 API：从入门到实战

Kafka 的事务 API 其实不复杂，核心就几个方法。我个人习惯把它们分成三组：初始化、事务边界、提交/回滚。

阶段	API 方法	作用
初始化	`initTransactions()`	初始化事务，注册生产者 ID
事务边界	`beginTransaction()`	开启一个新事务
提交/回滚	`commitTransaction()` / `abortTransaction()`	提交或终止当前事务
发送消息	`send()` + `sendOffsetsToTransaction()`	发送消息并提交消费位移

来看一个典型的代码骨架：

// 1. 配置事务属性
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("transactional.id", "txn-producer-001"); // 必须设置
props.put("enable.idempotence", "true");           // 幂等性必须开启

KafkaProducer<String, String> producer = new KafkaProducer<>(props);

// 2. 初始化事务
producer.initTransactions();

try {
    // 3. 开启事务
    producer.beginTransaction();

    // 4. 发送消息到主题A
    producer.send(new ProducerRecord<>("topic-a", "key1", "value1"));

    // 5. 发送消息到主题B（跨分区原子性）
    producer.send(new ProducerRecord<>("topic-b", "key2", "value2"));

    // 6. 提交消费位移（消费-生产模式）
    producer.sendOffsetsToTransaction(consumerOffsets, consumerGroupId);

    // 7. 提交事务
    producer.commitTransaction();
} catch (Exception e) {
    // 8. 回滚事务
    producer.abortTransaction();
} finally {
    producer.close();
}

避坑指南：我曾经在生产环境中忘记设置 transactional.id，结果事务一直报错。记住，这个 ID 必须唯一且稳定，Kafka 用它来恢复未完成的事务。

4.2 事务协调器：幕后的大管家

事务协调器（Transaction Coordinator）是 Kafka 事务的核心组件。你想想看，生产者怎么知道该找谁协调事务？答案是通过 transactional.id 的哈希值，找到对应的 Broker 分区。

具体流程是这样的：

生产者调用 initTransactions() 时，向任意 Broker 发送请求。
Broker 根据 transactional.id 的哈希值，确定负责的协调器分区。
协调器为该生产者分配一个 Producer ID (PID)，并递增 epoch。
后续所有事务操作，生产者都直接与这个协调器通信。

关键点：协调器负责维护事务状态（如 Ongoing、PrepareCommit、CompleteCommit），并记录到事务日志中。如果协调器挂了，新的协调器会从日志中恢复状态。

我记得有一次，集群中一个 Broker 宕机，导致部分事务协调器不可用。当时我以为是网络问题，查了半天才发现是协调器迁移导致的短暂不可用。嗯，这里要注意：协调器迁移期间，事务会阻塞，所以生产环境一定要做好监控。

4.3 事务日志：Kafka 自己的“数据库”

事务日志，说白了就是一个内部主题 __transaction_state。它记录了每个事务的元数据，包括：

事务 ID：对应生产者的 transactional.id
Producer ID 和 Epoch：用于幂等性和事务隔离
事务状态：Ongoing、PrepareCommit、CompleteCommit、Dead 等
参与的分区列表：事务涉及的所有 TopicPartition
最后心跳时间：用于超时判断

事务日志的写入是同步的，但读取是异步的。为什么这样设计？因为事务提交需要保证日志落盘，否则协调器挂了就恢复不了了。而读取状态时，可以容忍短暂的不一致。

性能陷阱：事务日志的副本数默认是 3，如果集群 Broker 数量不足 3，事务会一直报错。我曾经在测试环境只配了 2 个 Broker，结果事务提交一直超时，最后发现是副本数不够。

4.4 幂等性保证：从原理到实践

幂等性保证是 Kafka 事务的基础。你想想看，如果生产者重试导致消息重复，事务还怎么保证一致性？

Kafka 的幂等性通过 PID + Sequence Number 实现：

每个生产者有一个唯一的 Producer ID (PID)。
每个分区有一个 Sequence Number，从 0 开始递增。
Broker 会检查 Sequence Number，如果重复则拒绝写入。

举个例子：

// 开启幂等性
props.put("enable.idempotence", "true");

// 发送消息
producer.send(new ProducerRecord<>("topic", "key", "value"));

// 如果网络超时，生产者重试发送
// Broker 发现 Sequence Number 重复，直接返回成功

这里有个细节：幂等性只保证单个分区内的消息不重复，跨分区的事务一致性需要事务 API 来保证。所以，幂等性是事务的“地基”，但光有地基还不够。

个人经验：我建议在生产环境中始终开启幂等性，即使你不使用事务。因为它的性能开销很小（约 5%），但能避免很多数据重复的坑。

4.5 核心流程：一次完整的事务提交

为了让你更直观地理解，我画了一张流程图，展示一次完整的事务提交过程：

从图中可以看出，事务提交是一个两阶段提交的过程。协调器先写 PrepareCommit 日志，然后通知所有参与分区写入提交标记，最后写 CompleteCommit 日志。如果中间任何一步失败，协调器都会回滚事务。

4.6 常见问题与避坑

最后，分享几个我在项目中遇到的真实问题：

问题	原因	解决方案
事务超时	事务时间超过 `transaction.timeout.ms`（默认 1 分钟）	增大超时时间，或优化业务逻辑
协调器不可用	协调器所在 Broker 宕机或网络分区	确保集群有足够的副本，并配置监控告警
消息重复	未开启幂等性，或 `transactional.id` 重复	始终开启幂等性，确保 ID 唯一
事务日志积压	事务提交频率过高，日志清理不及时	调整 `log.retention.ms` 和 `log.segment.bytes`

重要提醒：我曾经在生产环境中遇到事务日志积压导致 Broker 磁盘写满的故障。原因是某个消费者一直不提交位移，导致事务一直处于 Ongoing 状态。所以，一定要监控 __transaction_state 主题的大小和事务状态分布。

好了，这一章的内容就到这里。Kafka 事务的核心就是协调器 + 日志 + 幂等性这三板斧。理解了它们，你就能在流式处理中做到“要么全部成功，要么全部失败”。