对于最新的稳定版本,请使用 Spring Integration 6.5.1! |
交易支持
从 5.0 版开始,新的TransactionHandleMessageAdvice
引入了使整个下游流事务化,这要归功于HandleMessageAdvice
实现。
当一个普通的TransactionInterceptor
用于<request-handler-advice-chain>
元素(例如,通过配置<tx:advice>
),已启动的事务仅适用于内部AbstractReplyProducingMessageHandler.handleRequestMessage()
并且不会传播到下游流。
要简化 XML 配置,以及<request-handler-advice-chain>
一个<transactional>
元素已添加到所有<outbound-gateway>
和<service-activator>
和相关组件。
以下示例显示<transactional>
使用中:
<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>
Java 配置可以通过使用TransactionInterceptorBuilder
,并且结果 Bean 名称可以在消息传递注释中使用 adviceChain
属性,如以下示例所示:
@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;
}
请注意true
参数TransactionInterceptorBuilder
构造 函数。
它会导致创建TransactionHandleMessageAdvice
,不是常规的TransactionInterceptor
.
Java DSL 支持Advice
通过.transactional()
选项,如以下示例所示:
@Bean
public IntegrationFlow updatingGatewayFlow() {
return f -> f
.handle(Jpa.updatingGateway(this.entityManagerFactory),
e -> e.transactional(true))
.channel(c -> c.queue("persistResults"));
}