如何向 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 开始的整个 sub-flow 的交易，并且 Propagation.NOT_SUPPORTED 会要求 SecondMessageTransformer 暂停当前交易并执行只有这个 MessageHandler 没有交易。使用 SecondMessageTransformer 完成工作后，原始事务应恢复并继续流程的其余部分。