java - 如何在 Spring Integration Java DSL 中处理 ExecutorChannel 的错误 channel

Question

我想知道如何限制仅调用者处理 Spring Integration Java DSL 中执行器 channel 的错误。

例如 在我的用例中，我有一个队列(大小为 100 以限制输入)供调用者发送事件，轮询器将轮询队列并发送到 ExecutorChannel 进行异步处理(假设处理是 CPU 密集型的，我们将任务执行器的池大小限制为 2)。异步处理的结果将发送回调用者。另外，如果异步处理抛出异常。原始调用者将处理异常，而不是让全局错误处理程序来处理它。

我不知道如何指定一个 channel ，只有调用者才能看到从 ExecutorChannel 抛出的错误，并在 Spring Integration Java DSL 中以私有(private)方式处理它。

Answer 1

为此，您必须为某些 MessageChannel 指定 errorChannel header ，这完全取决于您的调用者。

ExecutorChannel 基于 MessagePublishingErrorHandler，其代码如下:

private MessageChannel resolveErrorChannel(Throwable t) {
    Message<?> failedMessage = (t instanceof MessagingException) ?
            ((MessagingException) t).getFailedMessage() : null;
    if (this.defaultErrorChannel == null && this.channelResolver != null) {
        this.defaultErrorChannel = this.channelResolver.resolveDestination(
                IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME);
    }

    if (failedMessage == null || failedMessage.getHeaders().getErrorChannel() == null) {
        return this.defaultErrorChannel;
    }
    Object errorChannelHeader = failedMessage.getHeaders().getErrorChannel();
    if (errorChannelHeader instanceof MessageChannel) {
        return (MessageChannel) errorChannelHeader;
    }
    Assert.isInstanceOf(String.class, errorChannelHeader,
            "Unsupported error channel header type. Expected MessageChannel or String, but actual type is [" +
            errorChannelHeader.getClass() + "]");
    return this.channelResolver.resolveDestination((String) errorChannelHeader);
}

注意failedMessage.getHeaders().getErrorChannel()部分。

适合您的开箱即用的良好解决方案是带有 errorChannel 选项的 @MessagingGateway。这个使用完全相同的 errorChannel header 技术，并让处理或重新抛出异常完全在该 @MessagingGateway 的范围内。