如何向 Java DSL 集成流添加事务支持

Question

我必须向集成流添加事务支持。假设有 3 个变压器。第一个和第三个转换器应该在同一个事务中完成，但第二个不应该在任何事务中完成。因此，如果第三个转换器发生错误，则不应提交来自第一个和第三个转换器的所有更改，但应提交来自第二个转换器的更改。我怎样才能做到这一点？我尝试添加.transform(FirstMessageTransformer, e -> e.transactional(true))，但随后所有转换器都在事务中完成。我也尝试添加.transform(FirstMessageTransformer, e -> e.transactional(false))，但它似乎效果不佳，因为所有转换器都提交了更改，即使发生异常也是如此。

@Bean
public IntegrationFlow myMessageFromMessageAmqpInboundFlow() {
    return IntegrationFlows.from(myInboundChannel)
            .transform(FirstMessageTransformer)
            .transform(SecondMessageTransformer)
            .transform(ThirdMessageTransformer)
            .channel(anOutputChannel)
            .get();
}

Answer 1

试试这样：

        .transform(FirstMessageTransformer, e -> e.transactional(true))
        .transform(SecondMessageTransformer, 
                        e -> e.transactional(
                                     new TransactionInterceptorBuilder()
                                             .transactionManager(txManager)
                                             .propagation(Propagation.NOT_SUPPORTED)
                                             .build()))
        .transform(ThirdMessageTransformer)

这样，您将拥有一个以FirstMessageTransformer 开头的整个子流的事务，而Propagation.NOT_SUPPORTED 将让SecondMessageTransformer 暂停当前事务并仅在事务外执行此MessageHandler。完成SecondMessageTransformer 的工作后，原始事务应恢复并在流程的其余部分继续。