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

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

这是网关:

@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块，但它只会捕获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添加到网关，并将扩展头移到流程的第二行(异步)。如果在流程的同步部分中抛出异常，则仍然会阻塞 Controller 。

最佳答案

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

该请求消息获得一个唯一的临时回复 channel ，该 channel 作为replyChannel header 添加。即使网关具有显式的replyChannel，也可以将其简单地桥接到请求的replyChannel上，这就是网关将回复与请求相关联的方式。

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

由于您要将结果返回到网关，然后继续；网关交互已“花费”，并且没有任何东西会收到发送到replyChannel header 的消息(包括异常)。因此，您看到的日志消息。

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

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

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

编辑

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

至少要对网关做出一个响应，这一点很重要。您应该将错误 channel 留在pub/sub的一个分支上，而在第二个分支上进行更新。这样，第一回合上的异常将被抛出给调用方(如果您想在该网关上执行某些操作，例如路由到异常处理程序，则可以在该网关上添加errorChannel)。您只能在第二回合上更新 header ，以便其异常直接进入您的错误处理程序。

如果将网关上的exceptionChannel设置为ojit_code，则两条腿都将出现异常。