知識庫 / Spring WebFlux RSS 訂閱

Flux<String> flux = Flux.just("A", "B", "C")
  .flatMap(word -> {
    if (word.startsWith("A")) {
      return Flux.just(word + "1", word + "2", word + "3");
    } else {
      return Flux.just(word);
    }
  })
  .flatMap(word -> {
    if (word.startsWith("B")) {
      return Flux.just(word + "1", word + "2");
    } else {
      return Flux.just(word);
    }
  });

我們通過添加兩個條件轉換階段，創建了動態的分支，從而為反應式流中的每個項目提供了多個邏輯路徑。

現在，是時候驗證我們反應式流中的條件流：

StepVerifier.create(flux)
  .expectNext("A1")
  .expectNext("A2")
  .expectNext("A3")
  .expectNext("B1")
  .expectNext("B2")
  .expectNext("C")
  .verifyComplete();

太棒了！看來我們成功了。此外，我們可以使用 flatMapMany() 針對 Mono 發佈者進行類似用例。

6. 使用副作用運算符

在本節中，我們將探討如何在處理反應式流的同時，執行基於條件的同步操作。

6.1. 使用 doOnNext()

我們可以使用 `doOnNext()` 運算符，在每個 reactive stream 的項目上同步執行副作用操作。

讓我們先定義 `evenCounter` 變量來跟蹤 reactive stream 中偶數的數量：

AtomicInteger evenCounter = new AtomicInteger(0);

現在，讓我們創建一個整數流，並使用 doOnNext() 運算符將其與鏈式連接，以計數偶數：

Flux<Integer> flux = Flux.just(1, 2, 3, 4, 5, 6)
  .doOnNext(num -> {
  if (num % 2 == 0) {
    evenCounter.incrementAndGet();
  }
  });

我們已將操作添加到 if 塊中，從而使計數器的條件遞增得以實現。

接下來，必須驗證在從反應式流中處理每個項目後，evenCounter 的邏輯和狀態：

StepVerifier.create(flux)
  .expectNextMatches(num -> num == 1 && evenCounter.get() == 0)
  .expectNextMatches(num -> num == 2 && evenCounter.get() == 1)
  .expectNextMatches(num -> num == 3 && evenCounter.get() == 1)
  .expectNextMatches(num -> num == 4 && evenCounter.get() == 2)
  .expectNextMatches(num -> num == 5 && evenCounter.get() == 2)
  .expectNextMatches(num -> num == 6 && evenCounter.get() == 3)
  .verifyComplete();

太棒了！我們得到了預期的結果。

6.2. 使用 doOnComplete()

類似於地，我們還可以根據接收到反應式流信號的條件來關聯操作，例如在已發佈所有項後發送的完成信號。

讓我們首先初始化 done 標誌：

AtomicBoolean done = new AtomicBoolean(false);

現在，讓我們定義一個整數流，並使用 doOnComplete() 運算符設置 done 標誌為 true：

Flux<Integer> flux = Flux.just(1, 2, 3, 4, 5, 6)
  .doOnComplete(() -> done.set(true));

需要注意的是，完整信號只發送一次，因此副作用動作最多觸發一次。

此外，為了驗證副作用的條件執行，請在各個步驟中驗證 done 標誌。

StepVerifier.create(flux)
  .expectNextMatches(num -> num == 1 && !done.get())
  .expectNextMatches(num -> num == 2 && !done.get())
  .expectNextMatches(num -> num == 3 && !done.get())
  .expectNextMatches(num -> num == 4 && !done.get())
  .expectNextMatches(num -> num == 5 && !done.get())
  .expectNextMatches(num -> num == 6 && !done.get())
  .then(() -> Assertions.assertTrue(done.get()))
  .expectComplete()
  .verify();

太棒了！我們可以看到 done 標誌只有在所有項目成功發射後才被設置為 true。但是，需要注意的是，doOnComplete() 僅適用於 Flux 發佈者，而對於 Mono 發佈者，我們必須使用 doOnSuccess()。

對於這種條件數據訪問，我們可以使用 firstOnValue() 運算符

首先，讓我們定義兩個來源，即 source1 和 source2，它們的延遲分別為 200 毫秒和 10 毫秒。

Mono<String[]> source1 = Mono.defer(() -> Mono.just(new String[] { "val", "source1" })
  .delayElement(Duration.ofMillis(200)));
Mono<String[]> source2 = Mono.defer(() -> Mono.just(new String[] { "val", "source2" })
  .delayElement(Duration.ofMillis(10)));

接下來，我們使用 firstWithValue() 運算符，結合這兩個數據源，並利用框架的條件邏輯來處理數據訪問：

Mono<String[]> mono = Mono.firstWithValue(source1, source2);

最後，讓我們通過將生成的項與源數據進行比較，以較低的延遲進行驗證：

StepVerifier.create(mono)
  .expectNextMatches(item -> "val".equals(item[0]) && "source2".equals(item[1]))
  .verifyComplete();

太棒了！我們已經搞定了。此外，請注意，firstWithValue() 僅適用於 Mono 發佈器。

8. 使用 zip() 和 zipWhen()

在本節中，我們將學習如何利用條件流控制，使用 zip() 和 zipWhen() 運算符。

8.1. 使用 <em>zip()</em>

我們可以使用 `zip()` 運算符將來自多個來源的發射（emissions）組合起來。 此外，我們還可以利用其組合函數為數據處理添加條件邏輯。

下面我們來看如何使用它來確定緩存和數據庫中的值是否不一致：

首先，讓我們定義 dataFromDB 和 dataFromCache 發佈器，以模擬具有不同延遲和值的來源：

Mono<String> dataFromDB = Mono.defer(() -> Mono.just("db_val")
  .delayElement(Duration.ofMillis(200)));
Mono<String> dataFromCache = Mono.defer(() -> Mono.just("cache_val")
  .delayElement(Duration.ofMillis(10)));

現在，我們對其進行壓縮</em/>，並使用其組合函數來添加判斷緩存是否一致的條件：

Mono<String[]> mono = Mono.zip(dataFromDB, dataFromCache, 
  (dbValue, cacheValue) -> 
  new String[] { dbValue, dbValue.equals(cacheValue) ? "VALID_CACHE" : "INVALID_CACHE" });

StepVerifier.create(mono)
  .expectNextMatches(item -> "db_val".equals(item[0]) && "INVALID_CACHE".equals(item[1]))
  .verifyComplete();

int userId = 1;
Mono<String> userAgeCategory = Mono.defer(() -> Mono.just(userId))
  .zipWhen(id -> Mono.defer(() -> Mono.just(20)), (id, age) -> age >= 18 ? "ADULT" : "KID");

StepVerifier.create(userDetail)
  .expectNext("ADULT")
  .verifyComplete();

Mono<String> noUserDetail = Mono.empty()
  .zipWhen(id -> Mono.just(20), (id, age) -> age >= 18 ? "ADULT" : "KID");

StepVerifier.create(noUserDetail)
  .verifyComplete();