如何在 Spring 集成中重试对 Webflux.outboundgateway 的失败调用

Question

我在流 DSL 语法中定义了一个 spring 集成流。我的一位经纪人是 Webflux.outboundGateway。当无法访问远程 URI 时，将抛出异常并发送到 "errorChannel"。我正在尝试让流程重试，但到目前为止没有成功（永远不会重试呼叫）。这是我的配置：

@Bean
public IntegrationFlow retriableFlow() {
    return IntegrationFlows
            .from(...)
            .handle(
                    WebFlux.outboundGateway(m ->
                        UriComponentsBuilder.fromUriString(remoteGateway + "/foo/bar")
                                            .build()
                                            .toUri(), webClient)
                    .httpMethod(HttpMethod.POST)
                    .expectedResponseType(String.class)
                    .replyPayloadToFlux(true), e -> e.advice(retryAdvice())
            )
            // [ ... ]
            .get();
}
@Bean
public Advice retryAdvice() {
   RequestHandlerRetryAdvice advice = new RequestHandlerRetryAdvice();
   RetryTemplate retryTemplate = new RetryTemplate();
   ExponentialBackOffPolicy retryPolicy = new ExponentialBackOffPolicy();
   retryPolicy.setInitialInterval(1000);
   retryPolicy.setMaxInterval(20000);
   retryTemplate.setBackOffPolicy(retryPolicy);
   advice.setRetryTemplate(retryTemplate);
   return advice;
}

我应该使用不同于 RequestHandlerRetryAdvice 的东西吗？如果可以，应该是什么？

Answer 1

Webflux 根据定义是异步的，这意味着 Mono（回复）在请求 completes/fails 时异步满足，而不是在调用线程上。因此，该建议无济于事，因为请求的 "send" 部分总是成功的。

您将不得不通过错误通道上的流执行重试（分配在流开始附近的某处）。也许，一些 header 表示您重试了多少次。

ErrorMessage 具有属性 failedMessage 和 cause；您可以重新发送 failedMessage.

您可以关闭异步以便调用线程阻塞，但这确实违背了使用 WebFlux 的全部目的。

如何在 Spring 集成中重试对 Webflux.outboundgateway 的失败调用

How to retry a failed call to a Webflux.outboundgateway in Spring integration

spring-boot

spring-integration

spring-integration-dsl