如何将@Transactional与@KafkaListener一起使用？

Question

是否有可能将声明式交易管理（通过@Transactional）与@KafkaListener注释方法一起使用？例如，我想使用它来为每个侦听器定义单独的 tx 超时。我的设置如下：

交易管理器：

@Bean
@ConditionalOnBean(value = {HibernateTransactionManager.class})
public ChainedKafkaTransactionManager<Object, Object> chainedHibernateTm(KafkaTransactionManager<String, String> kafkaTransactionManager,
                                                                                 org.springframework.orm.hibernate5.HibernateTransactionManager hibernateTransactionManager) {
  return new ChainedKafkaTransactionManager<>(
    kafkaTransactionManager, 
    hibernateTransactionManager);
}

卡夫卡监听器：

@KafkaListener(topic = "my_topic")
@Transactional(timeout = 5)
public void handleMessage(SomeMessage message){
}

问题是 - KafkaMessageListenerContainer 在调用此类方法之前创建自己的事务 - 它使用自己的 TransactionTemplate：

@Nullable
private TransactionTemplate determineTransactionTemplate() {
  return this.transactionManager != null
    ? new TransactionTemplate(this.transactionManager)
    : null;
}

未使用 TransactionInterceptor。那么如何为具体的 @KafkaListener 方法设置特定的 tx 超时呢？

Answer 1

通过 plain 进行事务管理的各个方面@Transactional并不像您想要的那样与 Kafka 侦听器一起工作 - 没有内置的交互。

有一个专门的章节如何在事务中使用应用程序事件：使用注释的事务绑定事件@TransactionalEventListener有助于确保事务已成功完成。

您可以使用 @EventListener 注解注册常规事件侦听器。如果需要将其绑定到事务，请使用@TransactionalEventListener。当您这样做时，监听器默认绑定到事务的提交阶段。