java - spring 集成流程中的错误处理实践

Question

我有一个 spring 集成流程，涉及异步执行、从网关向 Controller 返回值、返回值后继续集成流程。

这是网关:

@MessagingGateway
public interface GW {

    @Gateway(requestChannel = "f.input")
    Task input(Collection<MessengerIncomingRequest> messages);

}

流程如下:

@Bean
IntegrationFlow jFlow() {
        return IntegrationFlows.from(
        MessageChannels.executor("f.input", executor()))
        .split()
        .channel(MessageChannels.executor(executor()))
        .transform(transformer)
        .channel(routerChannel())
        .get();
}

@Bean
ThreadPoolTaskExecutor executor() {
        ThreadPoolTaskExecutor pool = new ThreadPoolTaskExecutor();
        ...
        return pool;
}

@Bean
MessageChannel routerChannel() {
        return MessageChannels
        .publishSubscribe("routerChannel", executor())
        .get();
}

@Bean
IntegrationFlow routerChannelFlow() {
        return IntegrationFlows
        .from(routerChannel())
        .publishSubscribeChannel(s -> s
        .subscribe(f -> f.bridge(null))
        .subscribe(process()))
        .get();
}

@Bean
IntegrationFlow process() {
        return f ->
        f.route(p -> p.getKind().name(),
        m -> m.suffix("Channel")
        .channelMapping(TaskKind.CREATE.name(), "create")
        ....
}

@Bean
IntegrationFlow createFlow() {
        return IntegrationFlows.from(
        MessageChannels.direct("createChannel"))
        .handle(routerService)
        .get();
}

如何为整个流程定义错误处理程序？最佳做法是什么？我知道我可以为网关方法调用放置一个 try/catch block ，但它只会捕获 jFlow 中发生的异常。 channel(routerChannel()) 之前的所有内容的流程.

如何处理其余流程的错误？还是整个流程？

更新

我为 publishSubscribeChannel 添加了错误处理程序

@Bean
IntegrationFlow routerChannelFlow() {
    return IntegrationFlows
            .from(routerChannel())
            .publishSubscribeChannel(s -> s
                    .subscribe(f -> f.bridge(null))
                    .subscribe(process())
                    .errorHandler(errorHandler))
            .get();
}

但这似乎没有帮助，因为如果出现异常，我会收到以下错误:

cMessagingTemplate$TemporaryReplyChannel : Reply message received but the receiving thread has already received a reply:ErrorMessage [payload=org.springframework.messaging.MessageHandlingException:

并且我的错误处理程序没有被调用。

更新

根据 Gary 的回答，我尝试了这段代码:

@Bean
IntegrationFlow jFLow() {
    return IntegrationFlows.from(
            MessageChannels.executor("f.input", executor()))
            .split()
            .channel(MessageChannels.executor(executor()))
            .transform(transformer)
            .channel(routerChannel())
            .get();
}

@Bean
IntegrationFlow exceptionOrErrorFlow() {
    return IntegrationFlows.from(
            MessageChannels.direct("exceptionChannel"))
            .handle(errorHandler, "handleError")
            .get();
}

    @Bean
MessageChannel exceptionChannel() {
    return MessageChannels.direct("exceptionChannel")
            .get();
}

@Bean
IntegrationFlow process() {
        return f ->
        f.enrichHeaders((spec) ->
                    spec.header("errorChannel", "exceptionChannel", true))
        f.route(p -> p.getKind().name(),
        m -> m.suffix("Channel")
        .channelMapping(TaskKind.CREATE.name(), "create")
        ....
}

@MessagingGateway(errorChannel = "exceptionChannel")

经过另一次编辑，我添加了 exceptionChannel到网关，并将丰富的 header 移动到我的流程的第二条腿(异步)。如果在流程的同步部分抛出异常， Controller 仍然会被阻塞。

Answer 1

首先，让我解释一下网关的工作原理 - 它应该有助于理解下面的解决方案。

请求消息获得一个唯一的临时回复 channel ，该 channel 被添加为 replyChannel header 。即使网关有一个明确的 replyChannel，它也只是桥接到请求的 replyChannel - 这就是网关将回复与请求相关联的方式。

现在，网关还将请求的 errorChannel header 设置为相同的回复 channel 。这样，即使流是异步的，也可以将异常路由回网关并抛给调用者或路由到网关的错误 channel (如果指定)。此路由由连接到 ErrorHandlingTaskExecutor 的 MessagePublishingErrorHandler 执行，后者包装了您的执行程序。

由于您要将结果返回给网关然后继续；该网关交互已“耗尽”，并且不会收到发送到 replyChannel header 的消息(包括异常)。因此，您看到的是日志消息。

因此，一种解决方案是修复发送到独立流的消息中的 errorChannel header 。使用 .enrichHeaders 替换(确保将 overwrite 设置为 true)由网关设置的 errorChannel header 。这应该在流程中尽快完成，以便任何异常都将被路由到该 channel (然后您可以在那里订阅您的错误处理程序)。

另一种解决方案是连接您自己的错误处理执行器，在其 MessagePublishingErrorHandler 上显式设置 defaultErrorChannel 并删除 errorChannel header 。

异步错误路由首先查找 header ；如果存在，则将错误消息路由到那里；如果没有 header 并且 MPEH 没有默认错误 channel ；消息将被路由到默认的 errorChannel，(通常)有一个 LoggingChannelAdapter 订阅。默认的 errorChannel 是一个发布/订阅 channel ，因此您可以订阅其他端点。

编辑

您正在更改发布/订阅之前的 channel 。

重要的是至少得到一个对网关的响应；您应该在发布/订阅的一条腿上单独留下错误 channel ，并在第二条腿上更新它。这样，第一条线路上的异常将被抛出给调用者(如果您想在那里采取一些行动，例如路由到您的异常处理程序，您可以向网关添加一个 errorChannel)。您必须只更新第二条路线的 header ，以便其异常直接进入您的错误处理程序。

如果您将网关上的 errorChannel 设置为您的 exceptionChannel，那么两条线路上的异常都会到达那里。