如需使用最新稳定版本，请使用 Spring Integration 7.0.4！spring-doc.cadn.net.cn

事务支持

从版本 5.0 开始，得益于 HandleMessageAdvice 的实现，引入了一个新的 TransactionHandleMessageAdvice，使整个下游流程具备事务性。当在 <request-handler-advice-chain> 元素中使用常规的 TransactionInterceptor 时（例如通过配置 <tx:advice>），启动的事务仅应用于内部的 AbstractReplyProducingMessageHandler.handleRequestMessage()，而不会传播到下游流程。spring-doc.cadn.net.cn

为了简化 XML 配置，除了 <request-handler-advice-chain> 之外，所有 <outbound-gateway> 和 <service-activator> 及相关组件中还添加了 <transactional> 元素。以下示例展示了 <transactional> 的使用：spring-doc.cadn.net.cn

<int-jdbc:outbound-gateway query="select * from things where id=:headers[id]">
        <int-jdbc:transactional/>
</int-jdbc:outbound-gateway>

<bean id="transactionManager" class="org.mockito.Mockito" factory-method="mock">
    <constructor-arg value="org.springframework.transaction.TransactionManager"/>
</bean>

如果您熟悉 JPA 集成组件，这样的配置并不陌生，但现在我们可以从流程中的任何位置启动事务——而不仅仅局限于 <poller> 或消息驱动通道适配器（例如 JMS）。spring-doc.cadn.net.cn

通过使用 TransactionInterceptorBuilder 可以简化 Java 配置，并且结果 Bean 名称可用于消息注解的 adviceChain 属性，如下示例所示：spring-doc.cadn.net.cn

@Bean
public ConcurrentMetadataStore store() {
    return new SimpleMetadataStore(hazelcastInstance()
                       .getMap("idempotentReceiverMetadataStore"));
}

@Bean
public IdempotentReceiverInterceptor idempotentReceiverInterceptor() {
    return new IdempotentReceiverInterceptor(
            new MetadataStoreSelector(
                    message -> message.getPayload().toString(),
                    message -> message.getPayload().toString().toUpperCase(), store()));
}

@Bean
public TransactionInterceptor transactionInterceptor() {
    return new TransactionInterceptorBuilder(true)
                .transactionManager(this.transactionManager)
                .isolation(Isolation.READ_COMMITTED)
                .propagation(Propagation.REQUIRES_NEW)
                .build();
}

@Bean
@org.springframework.integration.annotation.Transformer(inputChannel = "input",
         outputChannel = "output",
         adviceChain = { "idempotentReceiverInterceptor",
                 "transactionInterceptor" })
public Transformer transformer() {
    return message -> message;
}

请注意 TransactionInterceptorBuilder 构造函数上的 true 参数。它会导致创建 TransactionHandleMessageAdvice，而不是普通的 TransactionInterceptor。spring-doc.cadn.net.cn

Java DSL 通过端点配置中的 Advice 和 .transactional() 选项支持，如下示例所示：spring-doc.cadn.net.cn

@Bean
public IntegrationFlow updatingGatewayFlow() {
    return f -> f
        .handle(Jpa.updatingGateway(this.entityManagerFactory),
                e -> e.transactional(true))
        .channel(c -> c.queue("persistResults"));
}