Spring Integration DSL：触发子流并忽略响应的实现

初始配置 #

我是Spring Integration新手，当前配置了如下IntegrationFlow：

@Bean
IntegrationFlow incomingFlow(
    @Qualifier("auditTrailFlow") IntegrationFlow auditTrailFlow,
    MyTransform myTransform
) {
  return IntegrationFlow
      .from("channel.incoming")
      .route("headers['vendor']", mapping -> mapping
          .subFlowMapping("vendor1", subFlowMapping -> subFlowMapping
              .publishSubscribeChannel(messageTaskExecutor, pubSubConfig -> pubSubConfig
                  .subscribe(subFlow -> subFlow
                      .enrichHeaders(headerEnricherSpec -> {
                        headerEnricherSpec.header("action", "request");
                        headerEnricherSpec.header("extSys", "vendor1");
                      })
                      .to(auditTrailFlow))
                  .subscribe(subFlow -> subFlow
                      .gateway(searchTypeFlow())
                      .gateway("channel.send.message")
                      .transform(myTransform)
                      .publishSubscribeChannel(messageTaskExecutor, p -> p
                          .subscribe(s -> s
                              .enrichHeaders(headerEnricherSpec -> {
                                headerEnricherSpec.header("action", "response");
                                headerEnricherSpec.header("extSys", "vendor1");
                              })
                              .to(auditTrailFlow))
                          .subscribe(s -> s.<PayoffQuote>handle((payload, headers) -> payload))))))
          .subFlowMapping("vendor2", subFlowMapping -> subFlowMapping.to(otherFlow())))
      .get();
}

需求说明 #

• 希望触发第一个publishSubscribeChannel下的第一个订阅流，无需关心其返回值；第二个publishSubscribeChannel中的同类流同理，异步触发后继续主流程。
• 补充业务流程：
  1. 通过@RestController接收HTTP请求并转为ObjectRequest
  2. 触发对应流程返回ObjectResponse，流程需完成：
     • 异步保存ObjectRequest到数据库，无需等待响应
     • 转换ObjectRequest为ObjectMessage，发送JMS并接收响应得到ObjectMessageResponse
     • 转换ObjectMessageResponse为ObjectResponse返回给HTTP请求
     • 异步保存ObjectResponse到数据库，无需等待响应

最终实现方案 #

在Artem Bilan协助下的优化实现：

@Bean
IntegrationFlow incomingFlow() {
  return IntegrationFlow
      .from("channel.incoming")
      .wireTap(auditTrailFlow) // 这是一个独立的流，从执行器通道开始，异步将请求对象保存到数据库
      .gateway(searchTypeFlow)  // 独立流：将请求对象转换为消息对象
      .gateway("channel.jms")  // 独立流：发送/接收JMS消息，并转换为消息响应对象
      .route("headers['vendor']", mapping -> mapping
          .subFlowMapping("vendor1", subFlowMapping -> subFlowMapping
              .transform(myTransform)) // 将消息响应对象转换为HTTP请求等待的响应对象
          .subFlowMapping("vendor2", subFlowMapping -> subFlowMapping
              .transform(myTransform)) // 将消息响应对象转换为HTTP请求等待的响应对象
      )
      .bridge()
      .wireTap(auditTrailFlow) // 同上，异步将响应对象保存到数据库
      .get();
}

内容的提问来源于stack exchange，提问作者tlavarea