如何在 Spring Integration 中以分散 - 聚集模式的聚合（）方法之后调用子流答案

【问题标题】：How to call subflows after aggregate() method in scatter-gather pattern in Spring Integration如何在 Spring Integration 中以分散 - 聚集模式的聚合（）方法之后调用子流
【发布时间】：2022-08-14 01:32:11
【问题描述】：

在这里，我使用分散聚集模式。如果我想在aggregate() 之后和to() 之前调用另一个IntegrationFlow，我该怎么做？我可以在这里使用接收者流，以便我也可以使该流成为有条件的吗？

     @Bean
          public IntegrationFlow flow() {
            return flow ->
                flow.handle(validatorService, \"validateRequest\")
                    .split()
                    .channel(c -> c.executor(Executors.newCachedThreadPool()))
                    .scatterGather(
                        scatterer ->
                            scatterer
                                .applySequence(true)
                                .recipientFlow(flow1())
                                .recipientFlow(flow2())
                                .recipientFlow(flow3()),
                        gatherer ->
                            gatherer
                                .releaseLockBeforeSend(true)
                                .releaseStrategy(group -> group.size() == 2))
                    .aggregate(lionService.someMethod())
    // here I want to call other Integration flows
                    .gateway(someFlow())
                    .to(someFlow2());
          }

 @Bean
  public IntegrationFlow flow1() {
    return flow ->
        flow.channel(c -> c.executor(Executors.newCachedThreadPool()))
            .enrichHeaders(h -> h.errorChannel(\"flow1ErrorChannel\", true))
            .handle(cdRequestService, \"prepareCDRequestFromLoanRequest\");
  }
//same way I have flow2 and flow3, and I have set an custom error channel header for all the flows
 @Bean
  public IntegrationFlow someFlow() {
    return flow ->
        flow.filter(\"headers.sourceSystemCode.equals(\"001\")\").channel(c -> c.executor(Executors.newCachedThreadPool()))
            .enrichHeaders(h -> h.errorChannel(\"someFlow1ErrorChannel\", true))
            .handle( Http.outboundGateway(\"http://localhost:4444/test2\")
                .httpMethod(HttpMethod.POST)
                .expectedResponseType(String.class)).bridge();
  }

到目前为止，只要在任何流程中发生任何错误，它都会通过已分配给它们的自定义错误通道，然后我会处理错误，但是当我在 .gateway(someFlow()) 中使用 someFlow1() 时，该流程中发生的错误不是进入指定的错误通道。如何解决？

在 errorhandler 类中，我正在做类似下面的事情——

//errorhandlerclass

 @ServiceActivator(inputChannel = \"flow1ErrorChannel\")
  public Message<?> processDBError(MessagingException payload) {
    logger.atSevere().withStackTrace(StackSize.FULL).withCause(payload).log(
        Objects.requireNonNull(payload.getFailedMessage()).toString());
    MessageHeaders messageHeaders = Objects.requireNonNull(payload.getFailedMessage()).getHeaders();
    return MessageBuilder.withPayload(
            new LionException(ErrorCode.DATABASE_ERROR.getErrorData()))
        .setHeader(MessageHeaders.REPLY_CHANNEL, messageHeaders.get(\"originalErrorChannel\"))
        .build();
  }

 @ServiceActivator(inputChannel = \"someFlow1ErrorChannel\")
  public Message<?> processDBError(MessagingException payload) {
    logger.atSevere().withStackTrace(StackSize.FULL).withCause(payload).log(
        Objects.requireNonNull(payload.getFailedMessage()).toString());
    MessageHeaders messageHeaders = Objects.requireNonNull(payload.getFailedMessage()).getHeaders();
    return MessageBuilder.withPayload(
            new LionException(ErrorCode.CUSTOM_ERROR.getErrorData()))
        .setHeader(MessageHeaders.REPLY_CHANNEL, messageHeaders.get(\"originalErrorChannel\"))
        .build();
  }

同样，如果someFlow() 中有任何错误，则会显示错误，但我希望它转到我正在根据我的要求处理错误的方法。

另外，您可以看到我在someFlow() 中使用了过滤器，所以当过滤器表达式评估为真时没有问题，但当它变为假时，它会抛出错误，但我希望它转义并转到下一个，即@ 987654327@。我使用.bridge() 认为它会返回到以前的上下文，但那没有发生。我知道我的理解有些差距。请帮助解决以上两个问题。

标签： java spring-boot spring-integration spring-integration-dsl spring-integration-http

【解决方案1】：

要调用另一个流并返回主流，您可以使用gateway()。但这种流动最终必须回归。没有条件流之类的东西：您可以通过filter() 端点（或操作员，如果您愿意）发送到通道（流程中的下一个端点）或不发送。 to() 运算符是当前流中的终端，但您可以在该目标流中继续执行您想要的任何逻辑。看起来您需要花费一些时间来了解什么是消息通道以及它如何连接 Spring Integration 中的端点。 IntegrationFlow 只是表达业务任务的逻辑容器 - 在运行时它是它们之间的所有端点和通道。

【讨论】：

实际上，在接收流中，我们可以编写表达式来进一步确定是否执行该流，因此很想知道我们是否在 to() 或 gateway() 中有内容。但是根据您的建议，我知道我应该对这两个使用 filter() 。