出错时再次调用并在 webflux 中重试

Question

我想使用WebClientfrom spring webflux 执行以下操作：

• 称呼endpoint1

• 如果它失败并出现预期错误，那么

• 打电话endpoint2和

• 只重试endpoint1一次

我已经做到了这一点：

webclient.get()

  .uri("/endpoint1")

  .retrieve()

  .bodyToFlux(MyBody.class)

  .retry(error -> {

     if (error == expectedError) {

       webclient.get()

         .uri("/endpoint2")

         .retrieve().block();

       return true;

     } else {

       false;

     });

请求时我无法阻止，endpoint2因为我会收到以下错误：（block()/blockFirst()/blockLast() are blocking, which is not supported in thread我也不想阻止）。

也许我应该使用retryWhen，但我不确定如何使用它。

Answer 1

我完成这项工作的唯一方法是retryWhen我无法使用reactor.retry.Retry#doOnRetry，因为它只接受 aConsumer而不是 a Monoor Fluxor Publisher。片段如下：webclient.get()&nbsp; .uri("/endpoint1")&nbsp; .retrieve()&nbsp; .bodyToFlux(MyBody.class)&nbsp; .retryWhen(errorCurrentAttempt -> errorCurrentAttempt&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .flatMap(currentError -> Mono.subscriberContext().map(ctx -> Tuples.of(currentError, ctx)))&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .flatMap(tp -> {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; Context ctx = tp.getT2();&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; Throwable error = tp.getT1();&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; int maxAttempts = 3;&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; Integer rl = ctx.getOrDefault("retriesLeft", maxAttempts);&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; if (rl != null && rl > 0 && error == myExpectedError) {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; // Call endpoint and retry&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; return webclient.get()&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .uri("/endpoint2")&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .retrieve()&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .thenReturn(ctx.put("retriesLeft", rl - 1));&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; } else {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; // Finish retries&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; return Mono.<Object>error(error);&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; }));