错误处理如何将所有错误丢弃在多个线程中并将其发送到错误渠道

Question

我有两个线程，线程中的每个方法都会引发异常。如何获得每个线程中丢弃的所有错误？在此代码中，错误通道只会捕获其中一个错误。基本上，我的目标是捕获所有错误并将其发送给呼叫者（REST控制器）。任何帮助将不胜感激。谢谢。

Integration.java

public IntegrationFlow provisionUserFlow() {
return IntegrationFlows.from("input.channel")
  .publishSubscribeChannel(Executors.newCachedThreadPool(),
      s -> s
            .subscribe(f -> f.handle(provisionerA, "provision"))
            .subscribe(f -> f.handle(provisionerB, "provision"))
  .get();
}
@ServiceActivator( inputChannel = "errorChannel", outputChannel = "replyChannel")
public boolean processErrors(Exception message) throws RuntimeException{
System.out.println("Message" + message.getMessage());
System.out.println ("******************************");
throw new RuntimeException(message.getMessage());
}

mgateway.java

@MessagingGateway(errorChannel = "errorChannel")
public interface MGateway {
@Gateway(requestChannel = "input.channel", replyChannel = "replyChannel") 
boolean invokeProvisioner(User user);
}

解决方案

@Bean
public IntegrationFlow provisionUserFlow() {
return
    IntegrationFlows.from("input.channel")
    .publishSubscribeChannel(Executors.newCachedThreadPool(),
        s -> s.applySequence(true)
            .subscribe(f -> f.enrichHeaders(e -> e.header(MessageHeaders.ERROR_CHANNEL, "errorChannel", true))
                .handle(provisionerA, "provision")
                .channel("aggregatorChannel")
            )
            .subscribe(f -> f.enrichHeaders(e -> e.header(MessageHeaders.ERROR_CHANNEL, "errorChannel", true))
                .handle(provisionerB, "provision")
                .channel("aggregatorChannel"))
            )
        .get();
}
@Bean
public IntegrationFlow aggregateFlow() {
return IntegrationFlows.from("aggregatorChannel")
    .channel( aggregatorChannel)
    .aggregate( a -> a.processor( collect, "aggregatingMethod"))
    .get();
}
@Transformer( inputChannel = "errorChannel", outputChannel = "aggregatorChannel")
public Message<?> errorChannelHandler(ErrorMessage errorMessage) throws RuntimeException {
Message<?> failedMessage =  ((MessagingException) errorMessage.getPayload())
    .getFailedMessage();
Exception exception = (Exception) errorMessage.getPayload();
return  MessageBuilder.withPayload( exception.getMessage())
       .copyHeadersIfAbsent( failedMessage.getHeaders() )
       .build();
}

Answer 1

您看到@Gateway只是Java方法。它有一个返回，可能会引发一个例外。我仍然感到困惑，为什么人们认为春季融合起来有所不同。它完全基于java，没有做任何魔术 - 只称呼java方法。

现在让我们想象一下，如果您只使用RAW Java开发您的工作。是的，您将等待这两个线程依赖并构建单个返回到呼叫者。

我们可以与春季集成一样。只需使用Aggregator EIP即可。您可以在该错误通道中捕获错误消息，并通过其failedMessage s关联它们。.publishSubscribeChannel()可以提供该选项：

/**
 * Specify whether to apply the sequence number and size headers to the
 * messages prior to invoking the subscribed handlers. By default, this
 * value is <code>false</code> meaning that sequence headers will
 * <em>not</em> be applied. If planning to use an Aggregator downstream
 * with the default correlation and completion strategies, you should set
 * this flag to <code>true</code>.
 * @param applySequence true if the sequence information should be applied.
 */
public void setApplySequence(boolean applySequence) {

默认情况下是false。然后，聚合器只能依靠默认的correlationStrategy并收集一组错误，供您返回标题中的replyChannel。

您可以在参考手册中找到的所有信息：

https://docs.spring.io/spring-integration/docs/4.3.12.release/referene/reference/html/messaging-routing-routing-chapter-chapter.html#aggregator

https://docs.spring.io/spring-integration/docs/4.3.12.release/referene/reference/html/configuration.html#namespace-errorhandler