时间戳

首页 > 解决方案 > 使用 Spring WebFlux 的 webclient 在 Mono 上进行有条件的重复或重试

java - 使用 Spring WebFlux 的 webclient 在 Mono 上进行有条件的重复或重试

问题描述

我想要做的是使用 webclient 对 Webflux 中的 Mono 进行有条件的重复。情况如下：

我们有一些返回生成文档的业务休息服务服务。此文档的生成是由在此之前调用的另一个服务触发的。但是，回到正题：文档生成服务需要 10-30 秒。我们要做的是：在 10 秒后检查是否生成了文档（单声道）。如果是这样，一切都很好。如果没有，请在 5 秒后重复（或重试）并检查是否生成了文档。依此类推，直到（最坏情况）30 秒后超时。这可能吗？一些（伪）代码：

return this.webClient.post().uri(SERVICE_URL)).        
body(BodyInserters.fromObject(docRequest)).retrieve().
bodyToMono(Document.class).
delaySubscription(Duration.ofSeconds(10)).
repeat5TimesWithDynamicTimeDelayUntil(!document.isEmpty()).
subscribe();

格雷茨贝尔纳多

标签： javaspringwebclientspring-webfluxproject-reactor

解决方案

对的，这是可能的。

Mono有两个重新订阅的概念（因此，重新触发请求）

  • retry = 如果上游完成异常，则重新订阅
  • 如果上游成功完成， repeat = 重新订阅

每个概念都有Mono针对不同用例的多个重载方法。寻找retry*和repeat*方法。例如，要无延迟地重试最大次数，请使用retry(int numRetries).

retryWhen通过和方法支持更复杂的用例repeatWhen，如以下示例所示。

重试时

如果单声道以异常完成最多 5 次，每次尝试之间间隔 5 秒，则重试：

// From reactor-core >= v3.3.4.RELEASE
import reactor.util.retry.Retry;

this.webClient
        .post()
        .uri(SERVICE_URL)
        .body(BodyInserters.fromValue(docRequest))
        .retrieve()
        .bodyToMono(Document.class)
        .retryWhen(Retry.fixedDelay(5, Duration.ofSeconds(5)))
        .delaySubscription(Duration.ofSeconds(10))

重试构建器支持其他退避策略（例如指数）和其他选项以完全自定义重试。

请注意，retryWhen(Retry)上面使用的方法是在 reactor-core v3.3.4.RELEASE 中添加的，并且该retryWhen(Function)方法已被弃用。在 reactor-core v3.3.4.RELEASE 之前，您可以使用reactor-extras项目中的重试函数构建器来创建一个Function传递给retryWhen(Function).

重复时

如果您需要在成功时重复，请使用.repeatWhenor.repeatWhenEmpty而不是.retryWhen上面的。

使用reactor-extras项目中的repeat 函数构建器来创建repeat Function，如下所示：

// From reactor-extras
import reactor.retry.Repeat;

this.webClient
        .post()
        .uri(SERVICE_URL)
        .body(BodyInserters.fromValue(docRequest))
        .retrieve()
        .bodyToMono(Document.class)
        .filter(document -> !document.isEmpty())
        .repeatWhenEmpty(Repeat.onlyIf(repeatContext -> true)
                .exponentialBackoff(Duration.ofSeconds(5), Duration.ofSeconds(10))
                .timeout(Duration.ofSeconds(30)))
        .delaySubscription(Duration.ofSeconds(10))

如果您想在成功或失败时重新订阅，也可以将 a.retry*与 a链接。.repeat*

推荐阅读