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

Question

我有一个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()).

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

UPDATE

我添加了错误处理程序 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:

并且我的错误处理程序不会被调用.

UPDATE

根据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到网关,并将富集标题移动到我的流的第二段(异步).如果在流的同步部分抛出异常,则控制器被阻塞.

Answer 1

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

请求消息获得唯一的临时回复通道,该通道被添加为replyChannel标头.即使网关有一个显式的replyChannel,它只是桥接到请求replyChannel- 这就是网关如何将回复与请求相关联.

现在,网关还将请求的errorChannel标头设置为相同的回复通道.这样,即使流是异步的,也可以将异常路由回网关,并将其抛出到调用者或路由到网关的错误通道(如果指定).此路由由MessagePublishingErrorHandler连接到a的a ErrorHandlingTaskExecutor执行,包装执行程序.

因为您将结果返回到网关然后继续; 网关交互是"花费"的,没有任何东西会收到发送到replyChannel标头的消息(包括异常).因此,您看到的日志消息.

因此,一种解决方案是修复errorChannel发送到独立流的消息的标头.使用.enrichHeaders更换(请务必设置改写为真)errorChannel这是由网关设置标题.这应该在流程中尽快完成,因此任何异常都将路由到该通道(然后您可以在那里订阅您的错误处理程序).

另一种解决方案是连接您自己的错误处理执行程序,显式设置defaultErrorChannel它MessagePublishingErrorHandler并删除errorChannel标题.

异步错误路由首先查找标头; 如果存在,则错误消息在那里路由; 如果没有标题,MPEH没有默认的错误通道; 邮件将被路由到默认的errorChannel(通常)LoggingChannelAdapter订阅.默认errorChannel为发布/订阅者频道,因此您可以为其订阅其他端点.

编辑

您正在更改pub/sub之前的频道.

获得至少一个网关响应非常重要; 您应该将错误通道单独留在pub/sub的一条腿上,并在第二条腿上更新它.这样,第一条腿上的异常将被抛给调用者(errorChannel如果你想在那里采取一些动作,你可以添加到网关,例如路由到你的异常处理程序).您必须只更新第二个分支上的标题,以便它的异常直接发送到您的错误处理程序.

如果您将errorChannel网关设置为您的网关,exceptionChannel那么两条腿的例外情况将会在那里进行.