Spring WebFlux 中的重试-工具盒子

1、概览 {#1概览}

在云上构建分布式应用时，需要考虑到服务故障，这通常会涉及到重试。

Spring WebFlux 提供了一些失败后重试的工具。

本文将会带你了解如何在 Spring WebFlux 添加和配置重试功能。

2、用例 {#2用例}

本文使用一个 MockWebServer 来模拟外部系统暂时不可用，然后又变为可用的情况。

连接到 REST 服务的组件：

@Test
void givenExternalServiceReturnsError_whenGettingData_thenRetryAndReturnResponse() {

    mockExternalService.enqueue(new MockResponse()
      .setResponseCode(SERVICE_UNAVAILABLE.code()));
    mockExternalService.enqueue(new MockResponse()
      .setResponseCode(SERVICE_UNAVAILABLE.code()));
    mockExternalService.enqueue(new MockResponse()
      .setResponseCode(SERVICE_UNAVAILABLE.code()));
    mockExternalService.enqueue(new MockResponse()
      .setBody("stock data"));

    StepVerifier.create(externalConnector.getData("ABC"))
      .expectNextMatches(response -> response.equals("stock data"))
      .verifyComplete();

    verifyNumberOfGetRequests(4);
}

3、添加重试 {#3添加重试}

Mono 和 Flux API 中内置了两个关键的 retry 操作。

3.1、使用 `retry` {#31使用-retry}

retry 可以防止应用立即返回错误，并重新订阅指定次数：

public Mono<String> getData(String stockId) {
    return webClient.get()
        .uri(PATH_BY_ID, stockId)
        .retrieve()
        .bodyToMono(String.class)
        .retry(3);
}

无论 Web 客户端返回什么错误，都会重试最多三次。

3.2、使用 `retryWhen` {#32使用-retrywhen}

retryWhen 方法可用来创建一个可配置的重试策略：

public Mono<String> getData(String stockId) {
    return webClient.get()
        .uri(PATH_BY_ID, stockId)
        .retrieve()
        .bodyToMono(String.class)
        .retryWhen(Retry.max(3));
}

如上，可以通过设置一个 Retry 对象来描述所需的逻辑。

这里使用了 max 策略，指定最多重试次数。这与第一个示例相同，但这里可以有更多的配置选项。特别要注意的是，在这种情况下，每次重试都尽快发生。

4、添加延迟 {#4添加延迟}

无延迟重试的主要缺点是无法给故障服务恢复时间。这可能会使服务不堪重负，使问题更加严重，降低恢复的机会。

4.1、`fixedDelay` 重试 {#41fixeddelay-重试}

可以使用 fixedDelay 策略在每次重试之间添加延迟：

public Mono<String> getData(String stockId) {
    return webClient.get()
      .uri(PATH_BY_ID, stockId)
      .retrieve()
      .bodyToMono(String.class)
      .retryWhen(Retry.fixedDelay(3, Duration.ofSeconds(2)));
}

该配置允许两次重试之间有两秒钟的延迟，这可能会增加成功的几率。不过，如果服务器正在经历较长时间的中断，那么应该等待更长的时间。但是，如果将所有延迟都配置为很长的时间，那么短时间的中断将使服务更加缓慢。

4.2、`backoff` 重试 {#42backoff-重试}

可以使用 backoff 策略来代替固定间隔重试：

public Mono<String> getData(String stockId) {
    return webClient.get()
      .uri(PATH_BY_ID, stockId)
      .retrieve()
      .bodyToMono(String.class)
      .retryWhen(Retry.backoff(3, Duration.ofSeconds(2)));
}

如上，逐渐增加了两次重试之间的延迟 - 在本例中，大致是 2 秒、4 秒和 8 秒的间隔。这样，外部系统就有更好的机会从常见的连接问题中恢复，或处理堆积的任务。

4.3、`jitter` 重试 {#43jitter-重试}

backoff 策略的另一个好处是，它为计算出的延迟时间间隔增加了随机性或抖动。抖动可以帮助减少多个客户端同时重试导致的重试风暴。

默认情况下，该值设置为 0.5，相当于抖动最多为计算延迟的 50%。

使用 jitter 方法配置一个不同的值 0.75，以表示最多为计算延迟的 75% 的抖动：

public Mono<String> getData(String stockId) {
    return webClient.get()
      .uri(PATH_BY_ID, stockId)
      .accept(MediaType.APPLICATION_JSON)
      .retrieve()
      .bodyToMono(String.class)
      .retryWhen(Retry.backoff(3, Duration.ofSeconds(2)).jitter(0.75));
}

注意，可能的取值范围介于 0（无抖动）和 1（抖动最多为计算延迟的 100%）之间。

5、错误过滤 {#5错误过滤}

此时，来自服务的任何错误都会导致重试，包括 400 Bad Request 或 401 Unauthorized 等 4xx 错误。

显然，不应该在客户端出现此类错误时重试，因为再怎么重试，服务器的响应不会有任何不同。

因此，需要仅在特定错误的情况下应用重试策略。

首先，创建一个异常来表示服务器错误：

public class ServiceException extends RuntimeException {
    
    public ServiceException(String message, int statusCode) {
        super(message);
        this.statusCode = statusCode;
    }
}

接下来，创建一个包含 5xx 异常的错误 Mono，并使用 filter 方法来配置过滤策略：

public Mono<String> getData(String stockId) {
    return webClient.get()
      .uri(PATH_BY_ID, stockId)
      .retrieve()
      .onStatus(HttpStatus::is5xxServerError, 
          response -> Mono.error(new ServiceException("Server error", response.rawStatusCode())))
      .bodyToMono(String.class)
      .retryWhen(Retry.backoff(3, Duration.ofSeconds(5))
          .filter(throwable -> throwable instanceof ServiceException));
}

现在，只有在 WebClient Pipeline（管道）抛出 ServiceException 时重试。

6、处理重试次数耗尽的情况 {#6处理重试次数耗尽的情况}

最后，需要考虑所有重试都不成功的情况。

在这种情况下，策略的默认行为是传播一个 RetryExhaustedException，封装了最后一个错误。

可以通过使用 onRetryExhaustedThrow 方法并提供一个 ServiceException 的Generator（生成器）来覆盖这种行为：

public Mono<String> getData(String stockId) {
    return webClient.get()
      .uri(PATH_BY_ID, stockId)
      .retrieve()
      .onStatus(HttpStatus::is5xxServerError, response -> Mono.error(new ServiceException("Server error", response.rawStatusCode())))
      .bodyToMono(String.class)
      .retryWhen(Retry.backoff(3, Duration.ofSeconds(5))
          .filter(throwable -> throwable instanceof ServiceException)
          .onRetryExhaustedThrow((retryBackoffSpec, retrySignal) -> {
              throw new ServiceException("External Service failed to process after max retries", HttpStatus.SERVICE_UNAVAILABLE.value());
          }));
}

现在，请求会在一系列重试失败后出现 ServiceException 异常。