异步流中出站网关上的错误处理

Question

我有一个这样的集成流程：

@Bean
public IntegrationFlow inboundRequestFlow()
{
    return IntegrationFlows.from( inboundRequestGateway() )
        .log( ... )
        .filter(  requestValidator,
            spec -> spec.discardChannel( invalidRequestChannel() ) )
        .bridge( spec -> spec.requiresReply( false ) )
        .channel( asyncFlowChannel )
        .wireTap(
            //sends a NO_CONTENT reply if the request is ok
            flow -> flow.enrichHeaders( spec -> spec.header( HttpHeaders.STATUS_CODE, HttpStatus.NO_CONTENT ).defaultOverwrite( true ) )
                .transform( payload -> "" )
                .channel( inboundGatewayReplyChannel() )
        ).get();

}

它在 http 网关上接收请求，对其进行验证，如果一切正常，则将请求发送到“asyncFlowChannel”并以 204 回复入站网关。

'asyncFlowChannel' 是另一个在 Executor 通道上运行的 IntegrationFlow 的起点：

@Bean
public IntegrationFlow outboundFlow()
{
    return IntegrationFlows.from( asyncFlowChannel)
        .log( ... )
        .transform( ... )
        .transform( ... )
        .split(... )            
        .resequence( ... )
        .enrichHeaders( ... )
        .log( ... )
        .transform( ... )
        .handle( this.outboundSOAPGateway() )
        .log( .. )
        .handle( ... )
        .bridge( spec -> spec.requiresReply( false ) )
        .channel( anotherAsyncFlowChannel )
        .get();
}

如果我的 outboundGateway 发生异常（由于网络相关的 IO 错误或错误响应），我想记录错误并采取适当的措施。但我无法在 outboundSOAPGateway 上设置错误通道，并且起始流程上的 inboundRequestGateway 已经收到它的回复。

我得到的错误的唯一线索是这个日志：

10:19:53.002 WARN [outbound-flow-0] org.springframework.messaging.core.GenericMessagingTemplate$TemporaryReplyChannel - 收到回复消息但接收线程已经收到回复：ErrorMessage [payload=..., headers =...]

我的问题是：在异步流中处理出站网关上的错误的正确方法是什么，其中启动流的 inboundGateway 已经收到它的回复？

Answer 1

任何MessageHandler 端点都可以通过AbstractRequestHandlerAdvice 提供。其中之一是ExpressionEvaluatingRequestHandlerAdvice，您可以在其中捕获异常并将其发送到failureChannel：https://docs.spring.io/spring-integration/docs/5.0.0.RELEASE/reference/html/messaging-endpoints-chapter.html#expression-advice。

为此，.handle( this.outboundSOAPGateway() ) 可以提供第二个参数，例如：

.handle((GenericHandler<?>) (p, h) -> {
                    throw new RuntimeException("intentional");
                }, e -> e.advice(retryAdvice()))

在这种情况下，我使用

@Bean
public RequestHandlerRetryAdvice retryAdvice() {
    RequestHandlerRetryAdvice requestHandlerRetryAdvice = new RequestHandlerRetryAdvice();
    requestHandlerRetryAdvice.setRecoveryCallback(new ErrorMessageSendingRecoverer(recoveryChannel()));
    return requestHandlerRetryAdvice;
}

但同样适用于ExpressionEvaluatingRequestHandlerAdvice。

顺便说一句，retryAdvice() 也可以为您解决问题。查看其ErrorMessageSendingRecoverer。

Answer 2

用.channel( asyncFlowChannel )代替

.gateway(asyncFlowChannel, e -> e.errorChannel(...))