跳到主要内容
版本：5.0.1

事务性绑定器

DeepSeek V3 中英对照 Transactional Binder

通过设置 spring.cloud.stream.kafka.binder.transaction.transactionIdPrefix 为非空值（例如 tx-）来启用事务。在处理器应用程序中使用时，消费者将启动事务；在消费者线程上发送的任何记录都会参与同一事务。当监听器正常退出时，监听器容器会将偏移量发送到事务并提交它。所有使用 spring.cloud.stream.kafka.binder.transaction.producer.* 属性配置的生产者绑定都会使用一个通用的生产者工厂；单个绑定的 Kafka 生产者属性将被忽略。

important

事务不支持普通的绑定器重试（以及死信处理），因为重试将在原始事务中运行，而该事务可能被回滚，并且任何已发布的记录也将被回滚。当启用重试时（通用属性 maxAttempts 大于零），重试属性用于配置 DefaultAfterRollbackProcessor，以在容器级别启用重试。类似地，此功能不是在事务内发布死信记录，而是通过在主事务回滚后运行的 DefaultAfterRollbackProcessor 移至监听器容器。

若您希望在源应用程序中使用事务，或从任意线程进行仅生产者事务（例如 @Scheduled 方法），您必须获取对事务性生产者工厂的引用，并使用它定义一个 KafkaTransactionManager bean。

@Bean
public PlatformTransactionManager transactionManager(BinderFactory binders,
        @Value("${unique.tx.id.per.instance}") String txId) {

    ProducerFactory<byte[], byte[]> pf = ((KafkaMessageChannelBinder) binders.getBinder(null,
            MessageChannel.class)).getTransactionalProducerFactory();
    KafkaTransactionManager tm = new KafkaTransactionManager<>(pf);
    tm.setTransactionId(txId)
    return tm;
}

请注意，我们使用 BinderFactory 获取对 binder 的引用；当仅配置了一个 binder 时，第一个参数使用 null。如果配置了多个 binder，则使用 binder 名称来获取引用。一旦获得对 binder 的引用，我们就可以获取对 ProducerFactory 的引用并创建事务管理器。

然后，你将使用常规的Spring事务支持，例如 TransactionTemplate 或 @Transactional，例如：

public static class Sender {

    @Transactional
    public void doInTransaction(MessageChannel output, List<String> stuffToSend) {
        stuffToSend.forEach(stuff -> output.send(new GenericMessage<>(stuff)));
    }

}

若希望将仅生产者的事务与其他事务管理器的事务同步，请使用 ChainedTransactionManager。

important

如果你的应用部署了多个实例，每个实例都需要一个唯一的 transactionIdPrefix。

Kafka 事务中的异常重试行为​

配置事务回滚重试行为​

在处理 Kafka 事务中的消息时，你可以通过 defaultRetryable 属性和 retryableExceptions 映射来配置哪些异常在事务回滚后应进行重试。

默认重试行为​

DefaultAfterRollbackProcessor 决定了哪些异常会在事务回滚后触发重试。默认情况下，所有异常都会触发重试，但你可以修改此行为：

spring:
 cloud:
   stream:
     kafka:
       bindings:
         <binding-name>:
           consumer:
             defaultRetryable: false  # Change default to NOT retry exceptions

当 defaultRetryable 设置为 false 时，DefaultAfterRollbackProcessor 将配置为 defaultFalse(true)，这意味着除非明确配置为可重试，否则异常将不会重试。

异常特定配置​

如需进行细粒度控制，您可以为特定异常类型指定重试行为：

spring:
 cloud:
   stream:
     kafka:
       bindings:
         <binding-name>:
           consumer:
             retryableExceptions:
               java.lang.IllegalStateException: true    # Always retry this exception
               java.lang.IllegalArgumentException: false  # Never retry this exception

DefaultAfterRollbackProcessor 将使用 addRetryableExceptions() 处理标记为 true 的异常，使用 addNotRetryableExceptions() 处理标记为 false 的异常。这些针对特定异常的配置将优先于默认行为。

实现细节​

  • 使用事务时，仅可在 retryableExceptions 中配置异常类型（Exception 的子类）

  • 若指定了非 Exception 类型，将抛出 IllegalArgumentException

  • 仅当启用事务且禁用批处理模式时，才会配置 DefaultAfterRollbackProcessor

  • 此配置确保事务重试行为与非事务性重试处理保持一致