Kafka事务机制详解：实现Exactly Once语义的端到端落地

1. 这不是“保证不丢不重”的口号，而是 Kafka 里最硬核的事务落地能力

你可能在文档里见过“Exactly Once”这个词，被包装成一句轻飘飘的承诺：“Kafka 支持精确一次语义”。但如果你真在生产环境里用过 Kafka，尤其是做过跨 Topic 的 ETL、状态更新、或者和数据库做双写，就会发现——这句话背后藏着一整套精密协同的机制，稍有不慎，要么数据重复写入导致金额多扣了一次，要么消息被跳过造成状态丢失。这不是理论题，是每天都在真实业务里发生的故障现场。

我带过的几个项目里，有某电商订单履约系统，在升级消费链路时把旧版的“At Least Once + 去重表”换成 Kafka 的事务 API，结果上线第三天就发现库存扣减重复了两次；还有某金融风控平台，用 Kafka 做实时特征拼接后写入 ClickHouse，因事务配置漏掉isolation.level=read_committed，下游读到了未提交的中间态数据，触发了误拦截。这些都不是 Kafka 的 Bug，而是对 Exactly Once 语义的理解偏差——它从来不是开个开关就能生效的魔法，而是一整套生产者、Broker、消费者三方严格配合的协议约束，其中事务机制（Transaction API）是它的核心引擎。

这篇文章要讲的，就是 Kafka 中 Exactly Once 语义如何通过事务机制真正落地。它不讲抽象概念，不堆术语定义，而是从一个真实消费-处理-写入的闭环出发，拆解每一步发生了什么、为什么必须这么设计、参数怎么调、哪些地方最容易踩坑。你会看到：事务 ID 怎么生成才不会冲突；Producer 如何用initTransactions()和beginTransaction()切换状态；Broker 怎样用__transaction_state内部 Topic 存储元数据；Consumer 怎么靠read_committed隔离级别过滤掉未提交消息；以及最关键的——当网络抖动、Broker 故障、Producer 崩溃时，这套机制如何靠幂等+事务协调器+超时回滚三重保障守住语义边界。如果你正在设计高一致性要求的流式处理链路，或者刚被重复消费问题折磨得睡不着，这篇就是为你写的实操手册。

2. Exactly Once 不是 Kafka 单方面能决定的，它是一场三方协同的精密配合

很多人以为“Kafka 支持 Exactly Once”，就意味着只要 Producer 发一次，Broker 存一次，Consumer 拿一次，就万事大吉。这是最大的误解。Kafka 的 Exactly Once 语义，本质是端到端（end-to-end）的语义保证，它覆盖的是“Producer → Kafka → Consumer → 外部系统（如 DB）”这个完整链条。而 Kafka 自身只负责中间这一段——Producer 到 Broker、Broker 到 Consumer。要让整个链条达成 Exactly Once，必须满足三个前提条件，缺一不可：

第一，Producer 必须启用幂等性（enable.idempotence=true）并配合事务 API 使用。幂等性解决的是单个 Producer 实例内重试导致的重复写入问题，它依赖producer.id和序列号（sequence number）做去重。但幂等性本身不提供跨会话、跨实例的语义保障——如果 Producer 进程挂了重启，新实例的producer.id不同，老的序列号就失效了。这时候就需要事务机制来兜底：事务赋予 Producer 一个全局唯一的transactional.id，Broker 用它关联所有会话周期内的状态，即使 Producer 重启，只要transactional.id不变，协调器就能恢复上下文，继续处理未完成的事务。

第二，Consumer 必须设置isolation.level=read_committed。这是最容易被忽略的一环。默认的read_uncommitted会让 Consumer 读到所有已写入日志的消息，包括那些还在事务中、尚未提交的“脏数据”。比如一个事务正在写入 topic-A 和 topic-B，Consumer 在 topic-A 里读到了消息，但此时事务失败回滚，topic-B 没写成，那这条 topic-A 的消息就成了孤岛，后续处理必然出错。read_committed强制 Consumer 只读取COMMIT标记后的消息，Broker 会在内部为每个事务消息打上控制标记（Control Record），Consumer 端据此过滤，确保看到的永远是原子一致的状态快照。

第三，应用层逻辑必须是幂等的。Kafka 能保证消息不重复投递，但不能保证你的业务代码不重复执行。比如你收到一条“用户支付成功”消息，然后调用 DB 更新余额。如果 DB 更新本身不是幂等操作（比如UPDATE balance SET amount = amount + 100），而 Kafka 因网络原因重发了这条消息，你的代码又没做防重，就会导致金额多加一次。所以事务机制只是提供了“不重复投递”的基础设施，真正的 Exactly Once，必须由业务代码配合唯一键、状态机、或数据库的INSERT ... ON CONFLICT DO NOTHING等手段来最终落地。

这三点构成一个铁三角：Producer 用事务 ID 锁定上下文，Broker 用控制记录和隔离级别做数据可见性管控，Consumer 用隔离级别过滤未提交数据。任何一角塌掉，Exactly Once 就会退化成 At Least Once 或 At Most Once。我在某支付中台项目里就吃过亏——当时只开了 Producer 事务，忘了改 Consumer 的isolation.level，测试环境一切正常，上线后遇到一次 Broker 网络分区，部分 Consumer 读到了未提交的风控规则变更，导致一批交易被错误拦截。后来我们把这三个条件做成上线 checklist，每次发布前强制核对，再没出过类似问题。

3. 事务机制的核心组件：从 Producer 初始化到 Broker 协调器的全链路解析

Kafka 的事务机制不是黑盒，它由几个关键组件协同工作，每一环都有明确职责和状态流转。理解它们，才能知道为什么某些参数必须设、某些步骤不能跳。下面我按实际运行顺序，带你走一遍从 Producer 初始化事务，到消息写入，再到 Consumer 消费的完整链路。

3.1 Producer 的事务生命周期：init → begin → send → commit/abort

Producer 启动事务的第一步，不是直接发消息，而是调用initTransactions()。这一步看似简单，实则关键：它会向集群中的 Transaction Coordinator（事务协调器）发起注册请求，携带transactional.id。Coordinator 收到后，会检查该 ID 是否已有活跃事务（比如上次崩溃未清理），如果有，会先触发 abort；然后为本次会话分配一个producerId（注意，这不是client.id，而是 Broker 分配的唯一长整型 ID），并初始化事务元数据。这个过程是阻塞的，如果 Coordinator 不可用或网络不通，initTransactions()会一直重试直到超时（默认transaction.timeout.ms=60000）。我建议在服务启动时就调用它，而不是等到第一次发消息才触发，避免首条消息延迟过高。

接着是beginTransaction()。这一步不发网络请求，只是在 Producer 内部将状态从READY切换到IN_TRANSACTION，并开始为后续发送的消息打上事务标记。此时 Producer 还没真正写入任何数据，只是“准备好了”。

真正写入发生在send()调用时。和普通发送不同，事务模式下的send()会做两件事：一是把消息体正常写入目标 Topic 的 Partition 日志；二是向__transaction_state这个特殊的内部 Topic 写入一条 Control Record（控制记录），内容包含transactional.id、当前事务状态（BEGIN）、时间戳等。__transaction_state是 Kafka 的“事务账本”，它本身也是分区副本的，由 Coordinator 负责管理。所有事务元数据——谁在写、写了哪些分区、是否提交——都存在这里。这也是为什么 Kafka 要求__transaction_state的副本数必须大于 1（默认min.insync.replicas=2），否则 Coordinator 无法保证元数据不丢。

最后是commitTransaction()或abortTransaction()。commit会向__transaction_state写入COMMIT标记，并通知所有涉及的 Partition Leader 将对应消息标记为“已提交”；abort则写入ABORT，Leader 会丢弃这些消息（它们仍留在日志里，但 Consumer 读不到）。无论 commit 还是 abort，Producer 状态都会回到READY，等待下一次beginTransaction()。

提示：transaction.timeout.ms参数极其重要。它定义了从beginTransaction()开始，到必须commit或abort的最大时间窗口。如果业务处理耗时长（比如调用外部 HTTP 接口），必须把这个值设大，否则 Coordinator 会主动 abort 事务，导致消息丢失。但也不能设太大，否则故障时事务悬挂时间过长，影响集群资源回收。我们一般按最长业务链路耗时 × 1.5 来设，比如最长处理 30 秒，就设 45000。

3.2 Broker 的事务协调器（Transaction Coordinator）：事务状态的中央大脑

每个 Kafka 集群有且仅有一个 Transaction Coordinator，它不是一个独立进程，而是由某个 Broker 动态担任（具体由transaction.state.log.replication.factor和transaction.state.log.min.isr参数决定其选举规则）。Coordinator 的核心职责是：管理所有transactional.id的状态、持久化事务元数据、响应 Producer 的事务请求、与 Partition Leader 协同标记消息状态。

Coordinator 的状态存储在__transaction_stateTopic 中。这个 Topic 默认有 50 个分区（可配），每个分区对应一个transactional.id的哈希槽位，确保同一个 ID 的所有操作都路由到同一分区，实现状态局部性。Coordinator 会为每个活跃事务维护一个内存状态机：Empty → Ongoing → PrepareCommit/PrepareAbort → CompleteCommit/CompleteAbort。状态变更时，它先写__transaction_state日志（需 ISR 同步成功），再更新内存状态。如果 Coordinator 崩溃，新选上的 Coordinator 会从__transaction_state日志中重放所有事件，重建状态，保证事务连续性。

Partition Leader 的角色也很关键。当 Producer 发送事务消息时，Leader 不仅要写入数据日志，还要在内存中维护一个“未提交消息范围”的映射表（firstUnstableOffset）。Consumer 设置read_committed后，Leader 会根据这个表和__transaction_state中的 COMMIT/ABORT 标记，动态计算出当前可读的 offset 范围，过滤掉未提交或已中止的消息。这个过程完全透明，Consumer 感知不到，但它决定了你能看到什么。

注意：__transaction_state的配置直接影响事务可靠性。我们曾在一个小集群里把replication.factor设为 1，结果 Coordinator 所在 Broker 故障，__transaction_state分区不可用，所有事务请求超时，整个写入链路卡死。后来强制要求：transaction.state.log.replication.factor必须 ≥ 3，且min.insync.replicas至少为 2，确保即使一台机器宕机，元数据仍可写入。

3.3 Consumer 的隔离级别：read_committed 如何过滤“脏数据”

Consumer 端的isolation.level=read_committed是 Exactly Once 的最后一道闸门。它的实现原理并不复杂：Consumer 在拉取消息时，会同时向 Partition Leader 请求两样东西——数据消息列表，以及该 Partition 当前的lastStableOffset（LSO）。LSO 是一个特殊 offset，代表所有已提交事务消息的边界。Leader 计算 LSO 的规则是：取所有未完成事务中最小的firstUnstableOffset，即“最早那个还没提交的事务的起始位置”。Consumer 收到响应后，只会返回 offset ≤ LSO 的消息，自动截断后面的部分。

举个例子：假设一个 Partition 日志里有 10 条消息，offset 0~9。其中 offset 5~7 属于一个未提交的事务。那么 LSO 就是 5，Consumer 只能看到 offset 0~4 的消息。一旦该事务 commit，LSO 更新为 8，Consumer 下次拉取就能看到 0~7；如果 abort，LSO 保持 5，5~7 永远不会被读到。

这个机制的好处是零侵入：Consumer 不需要理解事务细节，只需信任 Leader 返回的 LSO。坏处是它带来一点延迟——Consumer 必须等事务结束，才能看到之前的消息。所以在低延迟场景（比如毫秒级风控），我们会权衡：如果业务能容忍极少量重复（比如用 DB 唯一键兜底），有时反而用read_uncommitted+ 应用层幂等更高效。但对资金类、订单类强一致性场景，read_committed是必选项。

4. 实操全流程：从代码配置到集群验证，手把手复现一个事务闭环

光说原理不够，下面我用一个真实的端到端案例，带你从代码、配置、到集群验证，完整跑通一个 Kafka 事务流程。场景是：一个订单服务，监听ordersTopic，处理后将结果写入order_fulfillmentTopic，并同步更新 MySQL 订单状态。我们要确保这个“读-处理-写”三步原子执行，不丢不重。

4.1 Producer 端：Java 客户端事务配置与编码

我们用 Kafka 官方 Java Client（3.5.x 版本），核心配置如下：

Properties props = new Properties(); props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-broker1:9092,kafka-broker2:9092"); props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer"); props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer"); // 必须开启幂等 props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true"); // 指定事务 ID，必须全局唯一且稳定（不能用 UUID） props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "order-fulfillment-producer-01"); // 事务超时时间，按业务最长耗时设 props.put(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, "90000"); // 重试次数，事务模式下建议设高些 props.put(ProducerConfig.RETRIES_CONFIG, "2147483647"); // Integer.MAX_VALUE // ACK 级别必须为 all，确保所有 ISR 副本写入 props.put(ProducerConfig.ACKS_CONFIG, "all"); KafkaProducer<String, String> producer = new KafkaProducer<>(props); // 关键：初始化事务，必须在 send 前调用 producer.initTransactions(); // 模拟处理一条订单消息 String orderId = "ORD-2024-001"; try { producer.beginTransaction(); // 步骤1：写入 order_fulfillment Topic producer.send(new ProducerRecord<>("order_fulfillment", orderId, "FULFILLED")); // 步骤2：这里可以调用其他服务，比如更新 MySQL updateOrderStatusInDB(orderId, "FULFILLED"); // 步骤3：提交事务 producer.commitTransaction(); System.out.println("Transaction committed for order: " + orderId); } catch (Exception e) { // 任何异常都触发 abort，防止消息残留 producer.abortTransaction(); System.err.println("Transaction aborted for order: " + orderId + ", error: " + e.getMessage()); }

这段代码有几个关键点必须注意：

• TRANSACTIONAL_ID_CONFIG不能用随机值（如UUID.randomUUID().toString()），否则每次重启都生成新 ID，Coordinator 无法关联历史状态。我们用服务名+实例编号（如order-fulfillment-producer-01），确保稳定性。
• updateOrderStatusInDB()必须在commitTransaction()之前执行。如果 DB 更新失败，abortTransaction()会回滚 Kafka 消息，保证两边一致。反之，如果先 commit Kafka 再更新 DB，DB 失败就无法挽回。
• RETRIES_CONFIG设为最大值，是因为事务模式下 Producer 会自动重试失败的send()，但前提是enable.idempotence=true已开启，否则重试会导致重复。

4.2 Consumer 端：安全消费的配置与陷阱规避

Consumer 配置同样关键：

Properties props = new Properties(); props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-broker1:9092,kafka-broker2:9092"); props.put(ConsumerConfig.GROUP_ID_CONFIG, "order-fulfillment-consumer-group"); props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer"); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer"); // 必须设为 read_committed！ props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed"); // 自动提交关掉，手动控制 offset props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"); // session 超时和 heartbeat 时间要合理，避免误 rebalance props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "45000"); props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, "15000"); KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props); consumer.subscribe(Collections.singletonList("orders")); while (true) { ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100)); for (ConsumerRecord<String, String> record : records) { // 此时 record.value() 一定是已提交的事务消息 processOrder(record.value()); } // 手动提交 offset，确保和业务处理绑定 consumer.commitSync(); }

这里有个经典陷阱：ISOLATION_LEVEL_CONFIG必须显式设置。很多开发者以为enable.auto.commit=true就够了，其实不然。auto.commit只控制 offset 提交，不影响消息可见性。如果不设read_committed，Consumer 依然会读到未提交的“脏数据”，只是 offset 提交了而已。

另一个重点是commitSync()的时机。我们把它放在processOrder()之后，确保业务逻辑执行成功才提交 offset。如果processOrder()抛异常，offset 不提交，下次重启会重新消费，这正是 Exactly Once 要求的“处理失败可重试”。但要注意，processOrder()本身必须是幂等的，否则重试会出问题。

4.3 集群级验证：用命令行工具确认事务状态与消息可见性

代码写完只是第一步，必须验证集群是否真的按预期工作。Kafka 自带的kafka-transactions.sh工具是你的最佳帮手。

首先，查看当前活跃事务：

# 查看所有事务状态 bin/kafka-transactions.sh --bootstrap-server kafka-broker1:9092 --list # 查看指定 transactional.id 的详细状态 bin/kafka-transactions.sh --bootstrap-server kafka-broker1:9092 \ --describe --transactional-id order-fulfillment-producer-01

正常输出应该包含State: Empty（空闲）或State: Ongoing（进行中），以及TopicPartitions列出它写入的分区。

其次，验证消息是否被正确过滤。我们用kafka-console-consumer.sh分别用两种隔离级别消费：

# 用 read_uncommitted 消费（能看到所有消息，包括未提交的） bin/kafka-console-consumer.sh --bootstrap-server kafka-broker1:9092 \ --topic order_fulfillment --from-beginning --isolation-level read_uncommitted # 用 read_committed 消费（只能看到已提交的） bin/kafka-console-consumer.sh --bootstrap-server kafka-broker1:9092 \ --topic order_fulfillment --from-beginning --isolation-level read_committed

你可以手动模拟一个未提交的事务：在 Producer 代码里注释掉commitTransaction()，只留beginTransaction()和send()，然后运行。此时read_uncommitted消费者会立刻看到消息，而read_committed消费者一直收不到，直到你手动调用commitTransaction()或超时 abort。这就是事务机制在起作用。

最后，检查__transaction_stateTopic 的健康度：

# 查看其分区状态 bin/kafka-topics.sh --bootstrap-server kafka-broker1:9092 --describe --topic __transaction_state

确保所有分区的Leader都在线，Replicas和Isr数量一致（比如都是3），且没有Under Replicated Partitions。如果有，说明事务元数据同步有问题，必须先修复。

5. 生产环境高频问题排查：从超时中断到 Coordinator 不可用的实战指南

即便配置正确，生产环境也会冒出各种意料之外的问题。下面是我整理的 5 个最高频、最棘手的事务相关故障，附带根因分析和速查解决方案。每一个都来自真实线上事故，不是教科书理论。

5.1 问题：Producer 报错ProducerFencedException，事务被强制中止

现象：Producer 日志频繁出现org.apache.kafka.common.errors.ProducerFencedException: There is a newer producer with the same transactionalId，随后所有send()失败。

根因：transactional.id冲突。最常见的原因是：多个 Producer 实例使用了相同的transactional.id。比如你在 Kubernetes 部署了 3 个副本，但配置文件里写死transactional.id=order-producer，导致 3 个实例争抢同一个 ID。Coordinator 认为这是“脑裂”，会踢掉旧实例，给新实例发 Fence，旧实例就抛出此异常。

解决方案：

• 方案一（推荐）：在transactional.id后追加唯一标识，如 Pod 名称或实例序号。Kubernetes 中可用 Downward API 注入HOSTNAME，配置成order-producer-${HOSTNAME}。
• 方案二：如果必须单实例，确保 Deployment 的replicas=1，并用podAntiAffinity防止调度到同一节点。
• 方案三：临时救急，调大transaction.timeout.ms，给 Coordinator 更多时间清理旧状态，但这只是掩耳盗铃，必须从根源解决 ID 冲突。

实操心得：我们曾用方案一，在transactional.id里加入env=prod和shard=01，变成order-producer-prod-01，彻底杜绝冲突。记住，transactional.id的设计原则是：同一业务逻辑、同一环境、同一分片，ID 相同；不同分片、不同环境、不同业务，ID 必须不同。

5.2 问题：Consumer 长时间收不到消息，poll()返回空

现象：Consumer 日志显示poll()调用正常，但records.count()总是 0，read_committed模式下消息“消失”了。

根因：lastStableOffset（LSO）卡住了。通常是因为某个 Producer 的事务长时间未提交或 abort，比如业务代码里有个死循环，或者调用外部服务超时未处理。Coordinator 会一直等待这个事务结束，LSO 就停在它的起始位置，后面所有消息都不可见。

排查步骤：

1. 先用kafka-transactions.sh --list查看是否有Ongoing状态的事务。
2. 如果有，用--describe查看它关联的 TopicPartitions。
3. 检查这些 Partition 的 Leader 是否健康（kafka-topics.sh --describe）。
4. 查看 Producer 日志，确认它是否卡在某个send()或commitTransaction()上。

解决方案：

• 立即定位并 kill 掉卡住的 Producer 进程。
• Coordinator 会在transaction.timeout.ms后自动 abort 该事务（前提是配置了transaction.state.log.min.isr满足，否则 abort 请求也发不出去）。
• 为防患未然，在 Producer 业务逻辑里加超时控制：用CompletableFuture.orTimeout()包裹updateOrderStatusInDB()，超时则主动abortTransaction()。

5.3 问题：__transaction_stateTopic 出现 Under Replicated Partitions

现象：kafka-topics.sh --describe显示__transaction_state的某些分区Isr数量 <Replicas数量，如Replicas: 1,2,3但Isr: 1,2。

根因：Coordinator 所在 Broker 或其副本同步慢。__transaction_state是事务的“心脏”，如果它的 ISR 缩小，Coordinator 就无法写入元数据，所有事务请求都会失败。

紧急处理：

• 第一时间检查 Broker 日志，搜索Failed to write transaction state或Not enough in-sync replicas。
• 查看磁盘 IO、网络带宽、JVM GC 是否异常。我们曾遇到一次，因为__transaction_state所在 Broker 的磁盘写满，导致所有事务 hang 住。
• 临时方案：增加transaction.state.log.min.isr到当前 ISR 数量（如从 2 改成 1），但这会降低可靠性，只能作为应急，修复后必须改回。

长期预防：

• 为__transaction_state单独规划磁盘，不与其他高吞吐 Topic 共享。
• 监控其UnderReplicatedPartitions指标，阈值设为 0，告警立即响应。
• 定期用kafka-dump-log.sh检查其日志段是否异常增长（正常情况下，它只存元数据，体积很小）。

5.4 问题：Consumer 切换read_committed后，offset 重置，重复消费大量历史消息

现象：Consumer 首次启动时设isolation.level=read_committed，poll()立刻拉取到几千条历史消息，而之前用read_uncommitted时只从最新 offset 开始。

根因：Consumer Group 的初始 offset 策略。auto.offset.reset=earliest时，Consumer 会从 Topic 最早 offset 开始读。但read_committed模式下，它会跳过所有未提交的消息，所以实际读到的“最早可读消息”可能比earliest晚很多。而如果之前 Group 已提交过 offset，现在切模式，Kafka 会从已提交的 offset 继续，但如果那个 offset 对应的是未提交事务，它就会卡住，直到超时后 fallback 到earliest。

解决方案：

• 启动前，先用kafka-consumer-groups.sh查看当前 Group 的 offset：

  bin/kafka-consumer-groups.sh --bootstrap-server kafka-broker1:9092 \ --group order-fulfillment-consumer-group --describe

• 如果CURRENT-OFFSET是-（空），说明没提交过，此时auto.offset.reset生效。确保它是latest，而不是earliest。
• 如果CURRENT-OFFSET有值，但你想从头消费，先 reset offset：

  bin/kafka-consumer-groups.sh --bootstrap-server kafka-broker1:9092 \ --group order-fulfillment-consumer-group --reset-offsets --to-earliest --execute --topic orders

• 更稳妥的做法：新上线read_committedConsumer 时，创建新 Group ID，避免和旧 Group 混淆。

5.5 问题：事务提交成功，但下游 DB 更新失败，如何保证最终一致？

现象：ProducercommitTransaction()返回成功，但updateOrderStatusInDB()因网络抖动失败，订单状态没更新，而 Kafka 消息已提交，无法 rollback。

根因：这是分布式事务的经典难题。Kafka 事务只管 Kafka 内部，DB 是外部系统，两者没有两阶段提交（2PC）集成。

行业通用解法（我们采用的）：

• 方案一：本地消息表（推荐）
  在订单 DB 里建一张local_message表，字段包括order_id,status,kafka_topic,kafka_partition,kafka_offset,status（pending/processed）。Producer 先在 DB 里插入一条pending状态的记录，再发 Kafka 消息。Consumer 处理成功后，更新 DB 记录为processed。另起一个定时任务，扫描pending超过 5 分钟的记录，重新投递 Kafka 或告警人工介入。这样 DB 和 Kafka 的状态通过本地表对齐。

• 方案二：事务性发件箱（Transactional Outbox）
  类似方案一，但用 Debezium 监听local_message表的变更日志，自动同步到 Kafka。优点是解耦，缺点是引入新组件。

• 方案三：SAGA 模式
  把“更新 DB”作为事务的最后一步，如果失败，发一条补偿消息（如order_status_update_failed），由另一个 Consumer 执行回滚操作（如把 Kafka 消息标记为 dead letter）。适合复杂业务流程。

我们最终选了方案一，因为它简单、可控、不依赖额外中间件，且 DB 事务天然支持 ACID，可靠性最高。关键点是：local_message表的插入和 Kafkasend()必须在同一个 DB 事务里，用 Spring 的@Transactional注解包裹，确保要么都成功，要么都失败。

6. 我在真实项目里总结的 3 条硬经验：少走三年弯路

写到这里，你已经掌握了 Kafka Exactly Once 的全部技术要点。但作为在 Kafka 集群上摸爬滚打十年的老兵，我想分享三条血泪换来的经验，它们不写在官方文档里，却能帮你避开 80% 的坑。

第一条：永远不要在事务中做耗时的 I/O 操作，哪怕它看起来很“安全”。
我见过太多人把 HTTP 调用、Redis 查询、甚至文件读写塞进beginTransaction()和commitTransaction()之间。理由往往是“这个接口很稳定，超时只有 200ms”。但现实是，网络抖动、DNS 解析失败、下游服务 GC 暂停，都可能让一次调用卡住 5 秒以上。而transaction.timeout.ms一旦超时，Coordinator 就 abort，你的消息就丢了。正确的做法是：把所有外部依赖移到事务外，用异步回调或状态机驱动。比如，send()Kafka 消息后，立即commitTransaction()，然后用另一个线程池去调用 HTTP 接口，成功则更新 DB 状态，失败则发告警。Kafka 只负责可靠投递，不负责业务执行。

第二条：监控__transaction_state的延迟，比监控 Producer 的request-latency更重要。
Producer 的延迟指标告诉你“发消息快不快”，但__transaction_state的写入延迟（可通过kafka_server_transaction_coordinator_metricsJMX 指标RequestHandlerAvgIdlePercent和UnderReplicatedPartitions综合判断）告诉你“事务能不能成立”。我们曾经在一个集群里，Producer 延迟平均 5ms，一切正常，但__transaction_state的 ISR 经常掉到 1，导致事务成功率只有 60%。根本原因是 Coordinator 所在 Broker 的磁盘 IO 饱和。所以，务必把__transaction_state的健康度纳入核心监控大盘，它的优先级应该和 ZooKeeper 连接数一样高。

第三条：Exactly Once 不是银弹，要敢于在合适的地方降级。
不是所有场景都需要强一致性。比如日志采集、用户行为埋点、监控指标上报，这些数据丢了或重复了，业务影响微乎其微。强行上事务，会增加 Producer 的 CPU 开销（序列号计算、控制记录写入）、Broker 的存储压力（__transaction_state日志）、Consumer 的延迟（LSO 计算）。我们的做法是：对资金、订单、库存等核心链路，必须用事务；对非核心链路，用enable.idempotence=true+acks=all+ 应用层幂等（如消息 ID 去重），性能提升 30%，故障率反而更低。技术选型的本质，是权衡，不是炫技。

最后说一句：Kafka 的事务机制，是工业级流处理的基石，但它不是用来炫技的玩具。每一次initTransactions()的调用，背后都是对一致性的郑重承诺；每一行read_committed的配置，都是对数据质量的无声宣誓。把它用好，需要的不仅是技术，更是敬畏。