我在项目中定义了以下集成流
///
public IntegrationFlow acarsEventFlow() {
return IntegrationFlows
//.from(Jms.messageDrivenChannelAdapter(this.acarsMqListener)) //Get Message from MQ
.from(org.springframework.integration.jms.dsl.Jms.messageDrivenChannelAdapter(
org.springframework.integration.jms.dsl.Jms.container(this.acarsMqConnectionFactory, this.acarsQueue)
.transactionManager(transactionManager(this.acarsMqConnectionFactory))
.get()))
.wireTap(ACARS_WIRE_TAP_CHNL)
.transform(agmTransformer, "parseXMLMessage") //
.handle(acarsProcessor, "pushRawMessage") // (1)Call web service to push the message payload and if it fails then don't commit the transaction and rollback the message
.transform(agmTransformer, "populateSmi")
.filter(acarsFilter,"filterMessageOnSmi") //
.transform(agmTransformer, "populateImi") //
.filter(acarsFilter,"filterMessageOnSmiImi") //
.transform(acarsProcessor,"processEvent") //
.publishSubscribeChannel(pubSub -> pubSub
.subscribe(flow -> flow
.bridge(e -> e.order(Ordered.HIGHEST_PRECEDENCE))
.enrichHeaders(h -> h.headerExpression(KafkaHeaders.MESSAGE_KEY, "payload.flightNbr")) //Add flight number as key
.transform("payload.message") // publish the transformed message
.handle(Kafka.outboundChannelAdapter(kafkaTemplate).topic(acarsKafkaTopic))) //publish to kafka
.subscribe(flow -> flow
.channel(UPDATE_DATA_STORE_CHNL)))
.get();
}
///
我正在从MQ那里收到一条消息,启动了交易管理器,以确保该消息已回滚,除非已处理。现在,在一个句柄方法之一中,#PusprawMessage()[请参阅注释(1)调用Web服务以在上面的片段中推动消息有效负载]我需要致电WebService。目前,我只是从处理程序内部调用Web服务 - Pusprawmessage()。介绍通讯网关来致电第三方网络服务是个好主意吗?如果我们介绍一个消息传递网关,那么如何确保在网络服务下降时将原始消息回滚?
可以像现在一样拥有它。另外,最好使用.gateway()
在该网络服务过程中执行一些子流量。只要在同一线程中完成所有操作,只要您仅使用直接频道就可以参与同一交易。因此,该子流中的任何错误都会导致交易回滚。
只要您使用gateway()
,您也可以将该Web服务过程作为 async 。无论如何,它将在当前线程中等待回复或错误。因此,交易将再次回滚。