引言

今年 3 月的一个凌晨，Kafka 集群做 3.8 → 3.9 的滚动升级，库存服务照常处理订单。

库存扣减的代码长这样：

@Transactional  // 数据库事务
public void deduct(String orderId, String skuId, int quantity) {
    stockMapper.deduct(skuId, quantity);          // ① DB 扣减并提交
    stockLogMapper.insert(orderId, skuId, quantity);

    kafkaTemplate.send("stock-deducted", orderId, event)  // ② 再发消息
            .whenComplete((result, ex) -> {
                if (ex != null) {
                    log.error("send failed, orderId={}", orderId, ex);
                }
            });
}

平时这套代码跑了两年，没出过事。但那天凌晨，Broker 滚动重启期间分区 Leader 切换，ISR 一度缩到 1。Broker 端配了 min.insync.replicas=2，写入被拒绝；客户端重试到 delivery.timeout.ms（120 秒）耗尽，消息最终发送失败。

代码里的回调只做了一件事：打日志。

第二天对账，1,842 笔订单的库存已经扣了，下游履约、积分、通知服务一条 stock-deducted 都没收到。用户看到的订单一直停在"待发货"，客服电话被打爆。

更隐蔽的是另一个团队的反向事故：他们先发消息、后提交 DB。某次数据库死锁导致事务回滚，但消息已经发出去了，下游仓库照着消息发了货——超卖 312 单。

两起事故拼在一起，结论很扎心：

acks=all、副本、ISR 解决的是"消息在 Kafka 内部别丢"（见《Kafka 消息不丢失实战》）。它们管不了数据库提交成功、消息却没发出去，也管不了消息发出去、数据库却回滚了。这不是 Kafka 的可靠性问题，是两次提交之间根本没有原子性。

那 Kafka 事务能解决吗？

一半能，一半不能。

• 它能保证"发往多个 Topic / 多个分区的消息"同生共死；
• 它能保证"消费 offset + 产出的消息"同生共死，从而在 Kafka 内部做到 Exactly-Once；
• 但它跨不过数据库那道墙。Kafka 事务不是 XA，不包含 DB、Redis、HTTP。

这篇文章就围绕这个边界展开：先把三大能力讲透，再给出 Spring Kafka 的两种实战写法，最后落到生产配置、监控和踩坑清单。

先放结论总表：

一句话破题：Kafka 事务是"Kafka 内部的分布式事务"——它能让多个 Topic 的写入像一张表一样原子，能让 offset 和产出消息同生共死，但它跨不过 DB；把 DB 也拉进来，要么用本地消息表（Outbox），要么接受"最终一致 + 幂等"。

一、先把问题拆开：扣库存和发消息，为什么天然不一致

1.1 一次支付，两次提交

库存服务扣减 + 发消息，本质是两个独立系统各自提交一次：

业务方法开始
   │
   ├── ① DB 事务：update stock ...  ──commit──▶ 库存已扣（不可逆）
   │
   ├── ② Kafka 发送：producer.send(...) ──ack──▶ 下游可见
   │
   └── 业务方法结束

危险窗口就在 ① 和 ② 之间：
   - ①成功、②失败  → 库存扣了，下游不知道（漏单）
   - ②成功、①回滚  → 下游知道了，库存没扣（超卖）

在单体数据库时代，这两个动作可以放进同一个本地事务。但 Kafka 是另一个系统，数据库的 BEGIN/COMMIT 管不到它。于是只有三种结局：

1. 两个都成功：正常；
2. 一个成功一个失败：数据不一致；
3. 两个都失败：重试或补偿。

我们想要的是"要么都成功，要么都失败"，也就是原子性。但跨系统的强原子性需要分布式事务协议（XA/2PC），而 Kafka 事务不是 XA——它是 Kafka 自己实现的一套事务协调机制，只覆盖 Kafka 内部的写入。

1.2 四种常见解法，先选对战场

这里要特别区分两个容易混淆的概念：

• RocketMQ 的"事务消息"：半消息 + 本地事务 + 回查，解决的是"本地事务和发消息"的最终一致；
• Kafka 的"事务消息"：一般指事务生产者 / consume-transform-produce，解决的是 Kafka 内部多分区写入和消费位点的原子性。

两者名字一样，机制完全不同。选型对比可以看《消息队列三巨头：Kafka vs RocketMQ vs Pulsar》。

回到开头的事故：如果"扣库存"和"发消息"在同一个服务里，Kafka 事务救不了你，正确解法是 Outbox（第五章会给完整链路）。但如果把库存扣减拆到下游、通过 Kafka 命令驱动，那么 Kafka 事务就能把"消费命令 + 产出结果事件"做成原子，DB 侧再用 orderId 幂等兜底——这就是本文的重点。

1.3 Kafka 事务的能力边界

记住一句话：Kafka 事务的边界就是 Kafka 的边界。

二、Kafka 事务的底层：Coordinator、两阶段、LSO

不理解底层，配置就会抄错。这一章只讲三样东西：谁在管事务、一次事务怎么走、消费者看到的"已提交"到底是什么意思。

2.1 三个角色

生产环境有一条硬要求：__transaction_state 的副本数必须和普通 Topic 一样可靠。

# Broker 端（生产环境）
transaction.state.log.replication.factor=3
transaction.state.log.min.isr=2

单副本的事务状态 Topic，等于把"消息是否提交"的判决权交给了一台机器。这台机器挂了，事务可能长时间无法恢复。

2.2 一次事务的完整时序

Producer                        Transaction Coordinator         数据分区
   │                                      │                       │
   │ ① initTransactions                   │                       │
   │ ─── FindCoordinator ────────────────▶│                       │
   │ ◀── InitProducerId(PID, epoch) ──────│                       │
   │                                      │                       │
   │ ② beginTransaction（本地状态）        │                       │
   │                                      │                       │
   │ ③ 第一次 send 到分区 P0               │                       │
   │ ─── AddPartitionsToTxn(P0) ─────────▶│                       │
   │                                      │                       │
   │ ④ send 记录 ────────────────────────────────────────────────▶│ 写入日志（未提交）
   │                                      │                       │
   │ ⑤ （消费-生产场景）                    │                       │
   │ ─── AddOffsetsToTxn ────────────────▶│                       │
   │ ─── TxnOffsetCommit ────────────────▶│                       │
   │                                      │                       │
   │ ⑥ commitTransaction                  │                       │
   │ ─── EndTxn(commit) ─────────────────▶│                       │
   │                                      │ ⑦ PREPARE_COMMIT 写事务日志
   │                                      │ ─── 写 COMMIT 控制记录 ─▶│
   │                                      │ ⑧ COMPLETE_COMMIT      │
   │ ◀── 成功 ────────────────────────────│                       │

关键点：

1. initTransactions() 只做一次，拿到 PID 和 epoch；Spring 会帮你做。
2. beginTransaction() 是客户端本地状态，不产生 RPC；真正的登记发生在第一次向某个分区发送时（AddPartitionsToTxn）。
3. 提交是同步的：commitTransaction() 要等 Coordinator 写完事务日志、并在所有涉及的分区写入 COMMIT 控制记录，才返回成功。
4. abort 也会写控制记录：ABORT 标记之后，read_committed 消费者会跳过这批记录。

这套机制常被叫做"两阶段"，但它不是 XA 的 prepare/commit——它依赖 Kafka 自己的事务日志和控制记录，不具备跨系统能力。

2.3 read_committed 到底过滤了什么：LSO

分区日志里不只有业务消息，还有控制记录（COMMIT / ABORT）。消费者开启 read_committed 后，只能读到 LSO（Last Stable Offset）之前的消息。

分区 P0 的日志：

offset:   0     1     2      3      4      5
        [c0]  [c1] [T2r0] [T2r1] [ctrl] [c5]
                       ▲
                     LSO = 2

read_committed：只能读到 0、1，poll 到 offset 2 就停住（不是丢消息，是等）
T2 commit 后：LSO 前进到 5，T2r0、T2r1 变为可见
T2 abort 后：LSO 也前进到 5，但 T2r0、T2r1 被跳过

LSO 的定义是：该分区上第一个未结束事务的起始 offset。只要有一个长事务没有提交/回滚，这个分区上所有消费者都会被卡在 LSO，读不到后面的消息——哪怕后面的消息属于其他已提交的事务。

这带来两个直接后果：

• 长事务会阻塞读。一个跑了 5 分钟还没提交的事务，会让整个分区的 read_committed 消费延迟 5 分钟；
• transaction.timeout.ms 是安全阀。默认 60 秒，超时后 Coordinator 会主动 abort，LSO 才能继续前进。

所以事务不是"包得越大越好"，批次大小要受 transaction.timeout.ms 和 max.poll.interval.ms 约束。

2.4 幂等生产者 vs 事务生产者

很多人以为开了 enable.idempotence=true 就有了 Exactly-Once，这是误解。

幂等生产者解决的是"重试别写出重复"，事务生产者解决的是"多次写入要么全成要么全败 + 跨会话隔离"。两者是叠加关系，不是替代关系。

生产者的基础配置仍然是《Kafka 消息不丢失实战》里那套：

acks=all
enable.idempotence=true
max.in.flight.requests.per.connection=5
retries=2147483647
delivery.timeout.ms=120000

事务只是在这套地基上，再加一层"原子提交"。

三、能力一：跨 Topic 原子写入

3.1 编程式：KafkaTemplate.executeInTransaction

适用场景：一个方法里要往多个 Topic 写消息，或者要按批提交。

@Service
public class OrderService {

    private final KafkaTemplate<String, Object> kafkaTemplate;

    public void pay(String orderId, String productId, int quantity) {
        OrderPaidEvent paid = new OrderPaidEvent(orderId, productId, quantity);
        InventoryDeductCommand command = new InventoryDeductCommand(orderId, productId, quantity);

        kafkaTemplate.executeInTransaction(operations -> {
            operations.send("order-events", orderId, paid);
            operations.send("inventory-commands", orderId, command);
            return true;   // 返回值会原样返回给调用方
        });
    }
}

这段代码的语义：

• 两条消息在同一个事务里，commit 后同时可见；
• 回调里抛出任何异常，事务 abort，两条消息都不可见；
• executeInTransaction 是本地事务，不参与外层全局事务（Spring 官方注释写得很明确）。

三个必须知道的细节：

1. 每次调用都会新建一个事务生产者。Spring Kafka 的实现是 createProducer(transactionIdPrefix) → initTransactions() → 业务回调 → commit/abort → close。所以不要把 executeInTransaction 放进 for 循环里逐条调用，固定开销会把吞吐打穿。
2. 不支持嵌套。源码里直接 assert：Nested calls to 'executeInTransaction' are not allowed。外层已经有 Kafka 事务时，内层再调会直接失败。
3. send 返回的 future 完成，不等于事务提交成功。事务里的 send 只是把记录放进本地缓冲区；真正的可见性由 commit 决定。业务判断成功与否，要看 executeInTransaction 是否正常返回。

3.2 声明式：@Transactional + KafkaTransactionManager

如果方法本身就是业务事务边界，用注解更自然：

@Service
public class OrderService {

    private final KafkaTemplate<String, Object> kafkaTemplate;

    @Transactional(
            transactionManager = "kafkaTransactionManager",
            rollbackFor = Exception.class)
    public void pay(String orderId, String productId, int quantity) {
        kafkaTemplate.send("order-events", orderId, new OrderPaidEvent(orderId, productId, quantity));
        kafkaTemplate.send("inventory-commands", orderId,
                new InventoryDeductCommand(orderId, productId, quantity));
    }
}

前提是容器里有 KafkaTransactionManager。Spring Boot 的自动装配条件很简单：只要配置了 spring.kafka.producer.transaction-id-prefix，就会创建这个 Bean。

spring:
  kafka:
    producer:
      transaction-id-prefix: tx-order-${INSTANCE_ID}

这里有个 90% 的人会踩的坑：

项目里同时有 DataSourceTransactionManager（Bean 名 transactionManager）和 KafkaTransactionManager（Bean 名 kafkaTransactionManager）。@Transactional 不写 transactionManager 时，默认按名字找 transactionManager——它会开启一个数据库事务，而 Kafka 的 send 完全不在事务里。消息立刻可见，回滚也回滚不掉。

所以只要项目里有 DB，就必须显式写 transactionManager = "kafkaTransactionManager"。这不是可选项。

另外几个老问题依然成立（详见《Transactional 失效的 8 个场景》）：

• 同类自调用：this.pay() 不走代理，注解失效；
• 异常被吞：方法内 catch 住异常不抛，事务不会回滚；
• 异步线程：@Async 或线程池里发消息，事务上下文不跟着线程走；
• 方法 private / final：代理拦不住。

3.3 两种写法怎么选

一个简单的判断：业务方法用 @Transactional，批处理/脚本用 executeInTransaction；两者不要嵌套。

3.4 验证：回滚后消息去哪了

部署一个最小验证，比看十遍文档都管用。下面这个场景在事务里发两条消息，第二条发送前抛异常：

@Transactional(transactionManager = "kafkaTransactionManager", rollbackFor = Exception.class)
public void payWithFailure(String orderId) {
    kafkaTemplate.send("order-events", orderId, paidEvent);
    kafkaTemplate.send("inventory-commands", orderId, command);
    throw new IllegalStateException("模拟第二个消息发送失败");
}

预期结果：

用命令行验证：

kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic order-events \
  --from-beginning \
  --property print.key=true \
  --isolation-level read_committed

把 --isolation-level 换成 read_uncommitted，就能看到"被回滚但确实写进日志"的记录。这个对比是理解 Kafka 事务最直观的实验。

生产环境不要用 read_uncommitted 做业务消费——它会把 abort 的消息也读出来，等于事务白做。

四、能力二：事务消息（消费-处理-生产原子）

这是 Kafka 事务最有价值的能力，也是"Exactly-Once"的地基。

4.1 先处理再提交 offset，为什么还会重复

普通消费者的流程：

poll 一批消息
  → 业务处理（写 DB、调接口）
  → 提交 offset

如果业务处理成功、提交 offset 失败（比如实例挂了、rebalance），消息会被重新投递——至少一次。这就是《消息重复消费排查》里的经典事故：钱多扣了一次。

那能不能先提交 offset 再处理？那样会变成至多一次：处理失败时消息已经"确认"了，直接丢。

所以普通消费者只能在"可能重复"和"可能丢失"之间二选一。Kafka 事务要做的，是把"处理结果"和"offset"绑在一起：

事务开始
  ├── 业务处理 + 产出消息（写入 Kafka）
  ├── 提交 offset（写进同一个事务）
  └── commit

commit 成功，offset 和产出消息一起生效；abort，两者一起回退。没有中间态。

4.2 把 offset 放进事务：sendOffsetsToTransaction

Kafka 客户端提供了 sendOffsetsToTransaction(offsets, groupMetadata)。这里的 groupMetadata 不是可选项：它携带消费组 generation 和 memberId，Coordinator 用它来判断"提交 offset 的消费者还是不是当前组的合法成员"。

kafkaTemplate.executeInTransaction(operations -> {
    // 1. 产出消息
    operations.send("inventory-events", command.orderId(), event);

    // 2. 把本批消费位点写进同一个事务
    Map<TopicPartition, OffsetAndMetadata> offsets = Map.of(
            new TopicPartition(record.topic(), record.partition()),
            new OffsetAndMetadata(record.offset() + 1));
    operations.sendOffsetsToTransaction(offsets, consumerGroupMetadata);
    return null;
});

groupMetadata 的作用是 fencing：如果消费者已经被踢出组（rebalance、超时、网络分区），旧实例的 offset 提交会被拒绝，避免"僵尸消费者"把位点往前推。

但手写这套很容易漏：offset 怎么算、批量里多个分区怎么处理、提交失败怎么重试。Spring Kafka 提供了容器级事务，把这些都封装了。

4.3 Spring Kafka 的容器级事务（推荐）

让监听容器开启事务，listener 里只管 send：

@Configuration
public class KafkaConfig {

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Object> inventoryListenerContainerFactory(
            ConsumerFactory<String, Object> consumerFactory,
            KafkaTransactionManager<?, ?> kafkaTransactionManager) {

        ConcurrentKafkaListenerContainerFactory<String, Object> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);

        // Spring Kafka 3.x 的写法：事务管理器放在 ContainerProperties 上
        factory.getContainerProperties().setKafkaAwareTransactionManager(kafkaTransactionManager);
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.RECORD);
        return factory;
    }
}

@Component
public class InventoryCommandConsumer {

    @KafkaListener(
            topics = "inventory-commands",
            groupId = "inventory-service",
            containerFactory = "inventoryListenerContainerFactory")
    public void onCommand(ConsumerRecord<String, InventoryDeductCommand> record) {
        InventoryDeductCommand command = record.value();

        // DB 扣减：独立事务，靠 orderId 幂等
        stockService.deductOnce(command);

        // 产出消息：与 offset 提交在同一个 Kafka 事务里
        kafkaTemplate.send("inventory-events", command.orderId(),
                new InventoryDeductedEvent(command.orderId(), true));
    }
}

容器会做这几件事：

1. listener 调用前开启 Kafka 事务；
2. listener 里的 send 自动加入这个事务（同一个 ProducerFactory）；
3. listener 正常返回 → 提交 offset + 提交消息；
4. listener 抛异常 → abort，offset 不前进，消息重新投递。

版本提醒：Spring Kafka 2.x 常见写法是 factory.setKafkaTransactionManager(...)；3.x 改成了 ContainerProperties#setKafkaAwareTransactionManager(...)。抄代码前先确认版本，别在 3.x 里找一个已经挪走的方法。

4.4 重平衡、僵尸消费者与毒消息

事务消费有三个必须处理的边界：

① 重平衡

事务期间发生 rebalance，消费者可能被踢出组。此时 offset 提交会失败，事务 abort，消息重新投递。这不是 Bug，是保证不丢的代价——消费端必须幂等。

② 事务时长与 max.poll.interval.ms

如果一次处理 + 事务提交超过 max.poll.interval.ms（默认 5 分钟），消费者会被判定失联，触发 rebalance；等它回来提交 offset，已经被 fence 掉了。

所以事务里不要做慢调用：DB 批量更新、HTTP 请求、大文件处理，全部放到事务外，或者拆小批次。

③ 毒消息

listener 抛异常 → abort → 重新投递 → 再抛 → 再投递。一个坏消息可以把整个分区卡死。

必须给事务消费配退出机制：

• 有限次重试（退避）；
• 超过次数转入死信 Topic（DLQ）；
• 业务异常分类：可重试（网络、锁冲突）vs 不可重试（参数非法、数据不存在）。

Spring Kafka 的 DefaultErrorHandler + DeadLetterPublishingRecoverer 可以完成这套，但要注意：DLQ 的发送不应该再放进原事务，否则"处理失败"和"发 DLQ"会互相纠缠。

4.5 事务消息的完整数据流

inventory-commands（命令 Topic）
        │
        ▼
┌─────────────────────────────────────────┐
│ 事务容器                                  │
│  begin                                   │
│   ├── DB 扣减（orderId 幂等，独立事务）    │
│   ├── send inventory-events              │
│   └── offset 提交（同一个 Kafka 事务）     │
│  commit / abort                          │
└─────────────────────────────────────────┘
        │
        ▼
inventory-events（结果 Topic）──▶ 履约 / 通知 / 审计

这条链路里：

• Kafka 内部：offset 和结果消息原子，不丢不重；
• DB 侧：deductOnce(orderId) 用唯一键幂等，重复投递不会重复扣；
• 下游：仍然要按 orderId 做幂等，因为"至少一次"的边界永远存在。

五、能力三：Exactly-Once 语义

5.1 EOS 的四个必要条件

很多人把 isolation.level=read_committed 当成 Exactly-Once 的全部，这是最大的误解。真正的 EOS 需要四个条件同时成立：

四个条件缺一不可。只开 read_committed 而不开事务生产者，读到的仍然是普通消息，没有任何 EOS 可言；只开幂等生产者，只能防重试重复，防不了跨分区和消费位点。

5.2 read_committed 的代价

read_committed 不是免费的：

• 读延迟增加：消费者要等 LSO 前进，长事务会直接拉高消费延迟；
• 吞吐下降：要跳过 abort 记录、解析控制记录；
• 长事务放大影响：一个分区上的长事务会阻塞该分区所有 read_committed 消费者。

这也是为什么事务批次不能无限大：

批次太小 → 每条一次 EndTxn，吞吐崩
批次太大 → 事务时长增加，LSO 阻塞变长，还可能撞 transaction.timeout.ms

经验区间是 100 ~ 500 条一个事务，再根据消息大小、Broker 负载和 transaction.timeout.ms 压测调整。

5.3 EOS 的边界：跨出 Kafka 就失效

这是全文最重要的一节。

假设库存扣减在 listener 里执行：

stockService.deductOnce(command);                  // ① DB 提交
kafkaTemplate.send("inventory-events", ...);       // ② Kafka 事务

如果 ① 成功、② 失败（事务 abort / 超时 / 实例挂掉），消息会重新投递，deductOnce 发现 orderId 已存在，直接返回上次结果——不会重复扣库存。

但这不是 Kafka 的 EOS，而是"至少一次 + 业务幂等"。

Kafka 事务能保证的是：

• 同一次事务里的多条消息原子；
• offset 和产出消息原子。

它保证不了：

• DB 提交和 Kafka 提交原子；
• Redis、HTTP 调用和 Kafka 提交原子；
• 下游服务一定只处理一次。

一旦事务里出现外部副作用，EOS 就退化成"至少一次 + 幂等"。这不是 Kafka 的缺陷，是分布式系统的现实：没有跨系统的原子提交，只有可重试的幂等操作。

5.4 外部副作用的三种正确姿势

库存扣减的幂等实现，核心就三行：

// 1. 先查去重表
if (deductLogMapper.exists(orderId)) {
    return deductLogMapper.get(orderId);      // 重复投递，返回上次结果
}

// 2. 扣库存
int updated = stockMapper.deduct(skuId, quantity);
if (updated == 0) {
    throw new InsufficientStockException();
}

// 3. 写去重记录（与扣减同一个 DB 事务）
deductLogMapper.insert(orderId, skuId, quantity);

唯一键是底线。没有唯一键，"先查后插"在并发下依然会重复。

5.5 一个完整的 EOS 判定表

结论：EOS 只在 Kafka 内部成立。跨出 Kafka，一律按"至少一次 + 幂等"设计。

六、性能账：事务不是免费的

6.1 成本模型

一次 Kafka 事务的固定开销：

所以事务的成本结构是"每事务一个固定价 + 每条消息一个边际价"。批次越小，固定价摊得越薄不了。

6.2 量级参考

下面这组是 3 Broker、acks=all、幂等开启、256B 消息的常见压测区间。只看比例，不看绝对值——不同磁盘、分区数、消息大小差异很大，必须用自己的集群复测。

两个结论：

1. 单条一事务是不可接受的。吞吐可能只剩个位数百分比，延迟还高；
2. 批量提交后，事务的代价可以接受。60% ~ 80% 的非事务吞吐，换来多分区原子和 EOS，对大多数业务是划算的。

6.3 自己压一遍

不用 JMH，一个 KafkaProducer 循环就够：

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.ACKS_CONFIG, "all");
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "bench-" + UUID.randomUUID());

int total = 100_000;
int batchSize = 200;

try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
    producer.initTransactions();
    long start = System.nanoTime();
    for (int sent = 0; sent < total; sent += batchSize) {
        producer.beginTransaction();
        for (int i = 0; i < batchSize && sent + i < total; i++) {
            producer.send(new ProducerRecord<>("tx-bench", "k-" + (sent + i), payload));
        }
        producer.commitTransaction();
    }
    long elapsedMs = (System.nanoTime() - start) / 1_000_000;
    System.out.printf("batch=%d, throughput=%.0f msg/s%n", batchSize, total * 1000.0 / elapsedMs);
}

把 batchSize 改成 1 / 50 / 200 / 500 各跑一遍，就能得到自己集群的曲线。重点看两个指标：

• 吞吐：msg/s；
• commit 延迟：commitTransaction() 的 p99，它决定了事务方法的响应时间上限。

6.4 调优清单

七、生产落地清单

7.1 配置

生产者

acks=all
enable.idempotence=true
max.in.flight.requests.per.connection=5
transaction.timeout.ms=60000
delivery.timeout.ms=120000
compression.type=zstd

transaction.timeout.ms 必须小于 Broker 的 transaction.max.timeout.ms，否则 initTransactions() 直接报 InvalidTxnTimeoutException。

Broker

transaction.state.log.replication.factor=3
transaction.state.log.min.isr=2
transaction.max.timeout.ms=900000
transaction.abort.timed.out.transaction.cleanup.interval.ms=10000
transaction.remove.expired.transaction.interval.ms=3600000

前两项是底线：事务状态 Topic 必须多副本。没有它，事务提交的判决会受单机故障影响。

消费者

isolation.level=read_committed
enable.auto.commit=false

事务容器会接管 offset 提交；普通消费者仍然建议手动提交。

Spring Boot

spring:
  kafka:
    producer:
      transaction-id-prefix: tx-order-${INSTANCE_ID}
    consumer:
      isolation-level: read_committed

transaction-id-prefix 的三条规则：

1. 每个应用实例唯一：多实例共用前缀会互相 fence，事务反复失败；
2. 重启后保持稳定：不要用 UUID.randomUUID()，否则旧实例的僵尸事务永远无法被 fence；
3. Kubernetes 用 StatefulSet 序号或注入的实例 ID，不要用 Pod 名（重启会变）。

7.2 监控与告警

注意：kafka-consumer-groups.sh --describe 只能看到 LAG，看不到 LSO。要判断"是不是事务把读卡住了"，需要 consumer 的 last-stable-offset 指标，或者用 Admin API 查 LSO。

7.3 降级与兜底

降级不是关掉事务，而是换一条一致性与可用性的平衡线。 消费端幂等永远保留，它是所有降级方案的底座。

7.4 上线检查表

•  transactional.id 前缀唯一且稳定，多实例不冲突
•  所有相关消费者 isolation.level=read_committed
•  外部副作用（DB/Redis/HTTP）有唯一键或状态机
•  transaction.timeout.ms < transaction.max.timeout.ms
•  __transaction_state 副本数 3、min ISR 2
•  事务批次压测过，commit p99 达标
•  事务方法没有自调用、没有吞异常、没有异步线程
•  事务内没有慢 RPC
•  毒消息有重试上限和 DLQ
•  abort rate、commit latency、LSO lag 已接入监控
•  做过故障注入：回滚后 read_committed 确实看不到消息
•  明确事务边界：不把整个订单流程包成一个超大事务

八、常见问题

8.1 Kafka 事务能保证"扣库存 + 发消息"原子吗？

不能。 Kafka 事务只覆盖 Kafka 内部的写入和 offset。数据库提交是另一个系统，Kafka 没有能力让它一起提交或回滚。

如果扣库和发消息在同一个服务里，正确解法是本地消息表（Outbox）：业务数据和消息记录写在同一个 DB 事务里，再由 relay 或 CDC 投递到 Kafka。Kafka 事务在这里帮不上忙。

如果扣库在下游、由 Kafka 命令驱动，那么 Kafka 事务可以保证"消费命令 + 产出结果事件"原子，DB 侧用 orderId 幂等兜底。

8.2 那开头的事故到底该怎么修？

分工况：

不要用 Kafka 事务去"包"数据库操作——包不住，只会多一层超时和一致性幻觉。

8.3 @Transactional 为什么不生效？

四个高频原因，按出现频率排序：

1. 用错了事务管理器：项目里同时有 DB 和 Kafka 事务管理器，没写 transactionManager = "kafkaTransactionManager"，结果开的是 DB 事务；
2. 同类自调用：this.pay() 不走代理；
3. 异常被吞：方法内 catch 后没有重新抛出，事务不回滚；
4. 异步/线程池：事务绑定线程，跨线程后上下文丢失。

排查顺序：先看日志里有没有 KafkaTransactionManager 的 begin/commit，再看方法是不是被代理，最后看异常有没有抛出去。

8.4 executeInTransaction 和 @Transactional 怎么选？

• 业务服务方法：@Transactional，事务边界清晰，异常语义自然；
• 批处理/工具方法/一个方法里多次 send：executeInTransaction；
• 不要嵌套：外层有 Kafka 事务时，内层 executeInTransaction 会直接报错；
• 不要逐条调用：每次调用都会新建事务生产者，固定成本很高。

8.5 Kafka 的事务消息和 RocketMQ 的事务消息是一回事吗？

不是。

• RocketMQ：半消息 + 本地事务 + 回查，解决"本地事务和发消息"的最终一致；
• Kafka：事务生产者 + Coordinator + 控制记录，解决 Kafka 内部多分区/offset 的原子性。

两者名字相同，机制和边界都不同，不要混着讲。

8.6 用了事务消息，消费者还会重复消费吗？

会。 EOS 只在 Kafka 内部成立。

以下情况仍会重复：

• 事务提交成功，但消费端处理外部系统（DB/HTTP）失败后重试；
• 消费者 rebalance，offset 未提交；
• 事务超时被 abort，消息重新投递；
• 下游服务自己的重试。

所以消费端幂等是必须的，不是可选的。

8.7 read_committed 为什么会"卡住"？

因为 LSO。一个分区上只要有一个未结束的事务，read_committed 消费者就只能读到该事务开始之前的位置。

排查：

1. 看是否有长事务（transaction.timeout.ms 是否过大）；
2. 看是否有生产者挂了但事务没结束（Coordinator 会在超时后 abort）；
3. 看 __transaction_state 是否健康；
4. 看 consumer 的 last-stable-offset 与分区末尾的差距。

8.8 transaction-id-prefix 怎么配？

三条规则：

• 每个实例唯一：tx-order-${INSTANCE_ID}；
• 重启后稳定：不要用随机 UUID；
• 多生产者：Spring 会在前缀后追加序号，保证同一实例内多个生产者不冲突。

配错的典型症状：日志里反复出现 ProducerFencedException，事务提交失败，消息重复或丢失。

8.9 事务消息能跨 Kafka 集群吗？

不能。 事务 Coordinator 和 __transaction_state 都是集群内的，事务边界不跨集群。

跨集群只能各自事务 + 对账/幂等，或者用 MirrorMaker 做异步复制（复制的是已提交消息）。

8.10 事务会让吞吐下降多少？

看批次。单条一事务可能只剩个位数百分比；100 ~ 500 条一批，通常能保留 60% ~ 90% 的非事务吞吐。

优化优先级：批量提交 > 减少事务内操作 > 压缩 > 分区规划。不要为了事务把 acks 降回 1，那是拿可靠性换性能。

九、总结

选型决策树

你要的"原子"覆盖什么？
├── 只在 Kafka 内（多个 Topic / 多个分区）
│   ├── 一次性发送 → Kafka 事务（@Transactional / executeInTransaction）
│   └── 消费后再生产 → 事务容器 + offset 进事务 + read_committed
├── 数据库 + Kafka
│   ├── 能改架构 → Outbox（本地消息表）+ relay / CDC
│   └── 不能改 → 最大努力通知 + 对账 + 幂等
└── 多个外部系统（DB + Redis + HTTP）
    └── 没有银弹：状态机 + 幂等 + 补偿 + 对账

三大能力速查表

一句话

Kafka 事务的本质，是把 Kafka 内部的多次写入和消费位点打包成一个原子单元——它能让两个 Topic 的消息同生共死，让 offset 和产出消息同生共死，但它跨不过 DB 那道墙；把它当"分布式事务银弹"，第一次扣库存事故就会教你做人；把它用在正确的边界内（Kafka → Kafka 的原子与 EOS），再给外部副作用配上幂等和 Outbox，才是这套机制的正确打开方式。

关键结论

• 开头事故的根因不是"Kafka 丢消息"，是 DB 提交与 Kafka 发送之间没有原子性；
• acks=all + ISR 解决 Kafka 内部可靠，解决不了跨系统一致性；
• Kafka 事务不是 XA，不能覆盖 DB、Redis、HTTP、跨集群；
• EOS 需要四个条件同时成立：幂等生产者、事务、offset 进事务、read_committed；
• read_committed 的代价是 LSO，长事务会阻塞整个分区的消费；
• 事务的固定成本主要在 EndTxn、AddPartitionsToTxn 和每事务新建生产者，批量提交是第一位优化；
• transaction-id-prefix 必须实例唯一、重启稳定，否则 fencing 会互相踢；
• 消费端幂等是所有方案（事务、Outbox、降级）的共同底座。

给团队的建议

互动话题：你们项目里"DB 成功、消息失败"最后是怎么兜的？是本地消息表、对账补偿，还是干脆把扣减改成了消息驱动？有没有遇到过 read_committed 被长事务卡住、消费延迟突然飙高的情况？评论区聊聊，点赞最高的场景我拆成下一篇的完整事故复盘。

参考资料

• Apache Kafka Documentation: Transactions
• KIP-98: Exactly Once Delivery and Transactional Messaging
• KIP-447: Producer scalability for exactly once semantics
• Spring for Apache Kafka: Transactions
• Spring Boot: Kafka Properties
• Confluent: Exactly-once Semantics in Apache Kafka
• Confluent: Transactions in Apache Kafka
• 《Kafka 消息不丢失实战：生产者 ack + 副本 ISR + 消费者手动提交的完整配置》
• 《Kafka 顺序消息实战：同一个订单的消息必须按序消费》
• 《Kafka 为什么这么快？——顺序写 + 页缓存 + 零拷贝 + 批量》
• 《消息重复消费排查：幂等性缺失的血泪教训——钱多扣了一次》
• 《Transactional 失效的 8 个场景——90% 的 Java 开发者都踩过至少一个》
• 《消息队列三巨头：Kafka vs RocketMQ vs Pulsar——百万级消息吞吐实测》