事務支援
從 5.0 版本開始,引入了一個新的 TransactionHandleMessageAdvice
,藉助 HandleMessageAdvice
實現,它可以使整個下游流具有事務性。當在 <request-handler-advice-chain>
元素中使用常規的 TransactionInterceptor
時(例如,透過配置 <tx:advice>
),啟動的事務僅應用於內部的 AbstractReplyProducingMessageHandler.handleRequestMessage()
,並且不會傳播到下游流。
為了簡化 XML 配置,除了 <request-handler-advice-chain>
之外,所有 <outbound-gateway>
和 <service-activator>
以及相關元件中都添加了 <transactional>
元素。以下示例展示了 <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>
使用 TransactionInterceptorBuilder
可以簡化 Java 配置,並且其結果 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;
}
請注意 TransactionInterceptorBuilder
建構函式中的 true
引數。它會導致建立一個 TransactionHandleMessageAdvice
,而不是常規的 TransactionInterceptor
。
Java DSL 透過端點配置上的 .transactional()
選項支援 Advice
,如下例所示
@Bean
public IntegrationFlow updatingGatewayFlow() {
return f -> f
.handle(Jpa.updatingGateway(this.entityManagerFactory),
e -> e.transactional(true))
.channel(c -> c.queue("persistResults"));
}