Reactive Kafka中单个生产者多消费者能否实现Exactly-once语义？

1. 单生产者对应多消费者场景的 Exactly-once 可行性 #

完全可以实现，但需要满足核心前提：

• 生产者必须启用事务，通过Reactor Kafka的KafkaSender搭配TransactionManager管理事务生命周期
• 每个消费者的offset提交必须与生产者的消息发送绑定到同一个事务，确保「消费-处理-生产」全链路原子性
• 消费者需配置isolation.level=read_committed，避免读取未提交的事务消息，防止重复消费未确认的内容

2. 分布式消费者的响应式 Exactly-once 实现 #

分布式场景下同样可以通过响应式方式实现，关键在于事务边界的控制：

• 分布式消费者实例无需共享同一个TransactionManager，但每个实例的offset提交必须关联到生产者的事务上下文
• 利用Reactor Kafka的响应式API，手动控制offset提交时机：在消息处理完成且生产者发送成功后，通过receiverOffset.commitInTransaction(transactionManager)将offset提交纳入事务，保证事务要么全部成功，要么全部回滚
• 确保Kafka集群的事务日志（transaction.state.log）配置正确，保证事务状态在集群节点间同步，避免分布式环境下的事务状态不一致

3. 共享 TransactionManager 的方案 #

Reactor Kafka不兼容Spring Kafka的KafkaTransactionManager，但可以通过原生API实现共享事务管理：

• 基于全局的ProducerFactory创建单一TransactionManager实例，注入到所有需要的KafkaSender和消费者逻辑中
• 示例代码思路：

  // 全局ProducerFactory与TransactionManager
  ProducerFactory<String, String> producerFactory = new DefaultKafkaProducerFactory<>(producerConfigs());
  TransactionManager transactionManager = producerFactory.transactionManager();
  
  // 初始化KafkaSender
  KafkaSender<String, String> sender = KafkaSender.create(SenderOptions.create(producerConfigs()).producerFactory(producerFactory));
  
  // 消费者事务绑定逻辑
  receiver.receive()
      .flatMap(record -> {
          // 业务处理逻辑
          Mono<Void> processTask = handleRecord(record);
          // 发送消息+提交offset到事务
          return processTask
              .then(sender.send(Mono.just(SenderRecord.create(new ProducerRecord<>("output-topic", record.key(), record.value()), record.receiverOffset()))))
              .then(record.receiverOffset().commitInTransaction(transactionManager));
      })
      .subscribe();
  

• 注意事项：共享事务时，生产者的transactional.id必须全局唯一，多实例场景下可采用「前缀+实例ID」的方式生成，避免事务冲突

4. 与Spring Kafka事务组件的差异 #

Reactor Kafka的事务模型基于响应式流设计，和Spring Kafka的同步事务抽象（KafkaTransactionManager）不直接兼容。不建议强行适配，直接使用Reactor Kafka原生事务API更符合响应式编程模型，也能避免同步/异步模型冲突带来的问题

内容的提问来源于stack exchange，提问作者Yosi Pramajaya