Spring Cloud Stream函数式Consumer模型下@Transactional的使用方法

问题原因 #

Spring Cloud Stream 函数式消费模型默认未将Consumer执行逻辑、StreamBridge发送逻辑、数据库操作纳入同一个事务上下文，你单独在Consumer类加@Transactional只能触发默认数据源事务，RabbitMQ的消费确认、消息发送操作未和该事务绑定，因此抛出异常时消息发送仍会生效。

修复步骤 #

1. 新增事务绑定配置 #

修改application配置，将消费逻辑绑定到事务管理器，并开启RabbitMQ事务与Spring事务同步：

# 给输入绑定指定事务管理器，若使用链式事务则替换为对应Bean名称
spring.cloud.stream.bindings.functionalPayment-in-0.consumer.transaction-manager=transactionManager
# 开启RabbitMQ模板事务支持
spring.rabbitmq.template.transactional=true
# 日志生产者开启事务同步，仅当前事务提交后才真正发送消息
spring.cloud.stream.rabbit.bindings.log-out-0.producer.sync=true

2. （可选）配置多资源链式事务 #

如果需要严格保证数据库操作、RabbitMQ消费、RabbitMQ发送三者的事务一致性，可配置链式事务管理器：

import org.springframework.amqp.rabbit.transaction.RabbitTransactionManager;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.transaction.support.ChainedTransactionManager;

@Bean
public PlatformTransactionManager chainedTransactionManager(
        DataSourceTransactionManager dataSourceTransactionManager,
        RabbitTransactionManager rabbitTransactionManager) {
    // 注意顺序：后注册的事务先提交，避免Rabbit提交成功但数据库回滚的不一致问题
    return new ChainedTransactionManager(rabbitTransactionManager, dataSourceTransactionManager);
}

配置完成后将第一步配置中的transaction-manager值修改为chainedTransactionManager即可。

3. 调整Consumer代码 #

// 配置所有异常都触发回滚，避免非RuntimeException异常不回滚的问题
@Transactional(rollbackFor = Exception.class)
@RequiredArgsConstructor
@Component
public class FunctionalPayment implements Consumer<PendingPaymentEvent> {
    private final PaymentsService paymentsService;
    private final StreamBridge streamBridge;

    @Override
    public void accept(PendingPaymentEvent event) {
        paymentsService.triggerInvoicePayment(event.getInvoiceId());
        streamBridge.send("log-out-0", event);
        throw new RuntimeException("Test exception to rollback message from log-out-0");
    }
}

生效验证 #

配置完成后重新测试，抛出异常时会同时触发三类操作回滚：

• 数据库triggerInvoicePayment执行的变更回滚
• StreamBridge发送的日志消息不会入队
• 原消费消息不会被确认，会退回RabbitMQ队列（可配合重试、死信策略处理消费失败的消息）

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