Spring Kafka 3.2中为监听器容器设置事务管理器的非弃用方案

在Spring Kafka 3.2版本中，ContainerProperties.setTransactionManager()方法已被弃用，官方推荐直接通过ConcurrentKafkaListenerContainerFactory本身的setTransactionManager()方法配置事务管理器，无需再通过容器属性间接设置。

替代原配置的代码示例 #

@Bean
public ConcurrentKafkaListenerContainerFactory<Long, PersonDto> inContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<Long, PersonDto> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(inConsumerFactory());
    // 直接通过工厂设置事务管理器，替代原getContainerProperties().setTransactionManager()
    factory.setTransactionManager(kafkaTransactionManager);
    return factory;
}

多容器（部分启用事务）的配置方案 #

如果需要同时创建启用事务和不启用事务的监听器容器，只需分别定义多个ConcurrentKafkaListenerContainerFactory Bean即可：

1. 启用事务的容器工厂：

@Bean
public ConcurrentKafkaListenerContainerFactory<Long, PersonDto> transactionalContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<Long, PersonDto> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(transactionalConsumerFactory());
    factory.setTransactionManager(kafkaTransactionManager);
    return factory;
}

2. 不启用事务的容器工厂：

@Bean
public ConcurrentKafkaListenerContainerFactory<Long, PersonDto> nonTransactionalContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<Long, PersonDto> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(nonTransactionalConsumerFactory());
    // 不设置事务管理器，默认不启用事务
    return factory;
}

使用时，在@KafkaListener注解中指定对应的容器工厂：

// 使用带事务的容器工厂
@KafkaListener(topics = "transactional-topic", containerFactory = "transactionalContainerFactory")
public void handleTransactionalMessage(PersonDto dto) {
    // 事务内的业务逻辑
}

// 使用不带事务的容器工厂
@KafkaListener(topics = "non-transactional-topic", containerFactory = "nonTransactionalContainerFactory")
public void handleNonTransactionalMessage(PersonDto dto) {
    // 非事务的业务逻辑
}

关键说明 #

• 该变更的核心是将事务管理器的配置从容器属性层面提升到了工厂层面，更符合Spring容器配置的一致性逻辑。
• 每个容器工厂可以独立配置事务管理器、消费者工厂等属性，完全满足多场景的差异化需求，不受全局配置的限制。

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