知識庫 / Spring WebFlux RSS 訂閱

4. 使用 Flux.delayElements() 延遲元素

Flux 提供了一種更直接的方式來延遲其元素：

public class DelayElements {

    public static Flux<Integer> fetch(
      WebClient client, int requests, int delay) {
        return Flux.range(1, requests)
          .delayElements(Duration.ofMillis(delay))
          .flatMap(i -> RandomConsumer.get(client));
    }
}

使用 delayElements() 時，延遲直接應用於 Subscriber.onNext() 信號。換句話説，它會延遲 Flux.range() 中的每個元素。因此，傳遞給 flatMap() 的函數將受到影響，啓動時間會更長。例如，如果 delay 的值為 1000，則在我們的請求開始之前，會有一個一秒的延遲。

4.1. 調整我們的解決方案

由此可見，如果我們沒有提供足夠長的延遲時間，將會出現錯誤：

@Test
void givenSmallDelay_whenDelayElements_thenExceptionThrown() {
    int delay = 100;

    int requests = 10;
    assertThrows(InternalServerError.class, () -> {
      DelayElements.fetch(client, requests, delay)
        .blockLast();
    });
}

這是因為我們每請求等待 100 毫秒，但每個請求在服務器端完成需要兩秒。因此，我們的併發請求限制很快被達到，我們收到一個 500 錯誤。

我們如果添加足夠的延遲，可以規避請求限制。但是，那樣我們又會等待比必要時間更長。

根據我們的使用場景，等待時間過長可能會顯著影響性能。因此，接下來我們應該檢查一種更合適的處理方式，因為我們知道服務器的限制。

5. 使用 flatMap() 進行併發控制

鑑於我們服務的限制，我們的最佳方案是在最多 Concurrency.MAX_CONCURRENT 個請求並行發送。為此，我們可以將一個額外的參數添加到 <em >flatMap()</em> 中，用於指定最大並行處理數量：

public class LimitConcurrency {

    public static Flux<Integer> fetch(
      WebClient client, int requests, int concurrency) {
        return Flux.range(1, requests)
          .flatMap(i -> RandomConsumer.get(client), concurrency);
    }
}

此參數確保同時請求的最大數量不超過併發，並且我們的處理不會因不必要的原因而延遲：

@Test
void givenLimitInsideServerRange_whenLimitedConcurrency_thenNoExceptionThrown() {
    int limit = Concurrency.MAX_CONCURRENT;

    int requests = 10;
    assertDoesNotThrow(() -> {
      LimitConcurrency.fetch(client, TOTAL_REQUESTS, limit)
        .blockLast();
    });
}

當然，以下是翻譯後的內容：

不過，還有一些其他選項值得我們探討，具體取決於我們的場景和偏好。我們來詳細瞭解一下。

public class Resilience4jRateLimit {

    public static Flux<Integer> fetch(
      WebClient client, int requests, int concurrency, int interval) {
        RateLimiter limiter = RateLimiter.of("my-rate-limiter", RateLimiterConfig.custom()
          .limitRefreshPeriod(Duration.ofMillis(interval))
          .limitForPeriod(concurrency)
          .timeoutDuration(Duration.ofMillis(interval * concurrency))
          .build());

        // ...
    }
}

現在我們將其包含在我們的 Flux 中，使用 transformDeferred()，從而控制我們的 GET 請求速率：

return Flux.range(1, requests)
  .flatMap(i -> RandomConsumer.get(client)
    .transformDeferred(RateLimiterOperator.of(limiter))
  );

我們應該注意到，如果我們將時間間隔定義得過低，仍然可能存在問題。但是，這種方法在我們需要與其它操作共享速率限制規範時非常有用。

7. 精確的限速與 Guava

Guava 具有通用限速器，非常適合我們的場景。 此外，由於它使用令牌桶算法，因此只有在必要時才會阻止，而不是像 Flux.delayElements() 那樣每次都阻止。

首先，我們需要將 Guava 添加到我們的 pom.xml 中：

<dependency>
    <groupId>com.google.guava</groupId>
    <artifactId>guava</artifactId>
    <version>33.2.1-jre</version>
</dependency>

為了使用它，我們調用 <em >RateLimiter.create()</em > 並傳入我們希望發送的最大請求數每秒鐘。然後，在發送請求時，我們調用limiter的acquire()方法，以便在必要時進行請求限流：acquire()``。

public class GuavaRateLimit {

    public static Flux<Integer> fetch(
      WebClient client, int requests, int requestsPerSecond) {
        RateLimiter limiter = RateLimiter.create(requestsPerSecond);

        return Flux.range(1, requests)
          .flatMap(i -> {
            limiter.acquire();

            return RandomConsumer.get(client);
          });
    }
}

由於其簡潔性，該解決方案效果卓越——它不會使我們的代碼塊變得過長。例如，如果，出於某種原因，某個請求花費的時間超過預期，下一個請求將不會等待執行。但是，這種情況僅在我們在 requestsPerSecond 的範圍內時才會發生。

8. 結論

在本文中，我們探討了幾種用於限制 WebClient 速率的方法。隨後，我們模擬了一個受控的 Web 服務，以觀察其對我們代碼和測試的影響。此外，我們利用 Project Reactor 和一些庫，以不同的方式實現了相同的目標。