。

再看一遍前面的例子;保證第一個getPage()調用在針對每個附加頁面的后續調用之前發生。此外，由于后續對getPage()的調用是在.Flatmapmany()中的，所以由框架負責優化多線程執行，并將結果匯到一起返回，傳播可能發生的任何錯誤。

條件邏輯

與命令式編程不同，在響應式編程中錯誤是作為一種值來考慮的。這意味著它們是通過流操作來傳遞的。這些錯誤可以通過所有方式傳遞給消費者，或者流可以基于它們改變行為。這種行為變化可以表現為錯誤的轉換或基于錯誤產生新的結果。

public Mono<AppStatsResponse> getApplication(GetAppRequest request) {
  return client.applications()
    .statistics(AppStatsRequest.builder()
      .applicationId(request.id())
      .build())
    .onErrorResume(ExceptionUtils.statusCode(APP_STOPPED_ERROR),
      t -> Mono.just(AppStatsResponse.builder().build()));
}

在本例中，我們要求為正在運行的應用程序獲取統計信息。如果一切正常，響應就會傳回給消費者。但是，如果接收到一個錯誤(帶有特定的狀態代碼)，則返回一個空響應。使用者永遠不會看到錯誤和執行過程中的默認值，就好像從來沒有發出過錯誤信號一樣。

如前所述，一個流完成時未發送任何條目也是有效的。通常，這就相當于返回null(其中void返回類型是一種特殊情況)。像以上這種出錯的情況一樣，沒有任何條目的完成結果可以一直傳遞給消費者，或者流可以基于它們改變行為。

public Flux<GetDomainsResponse> getDomains(GetDomainsRequest request) {
  return requestPrivateDomains(request.getId())
    .switchIfEmpty(requestSharedDomains(request.getId()));
}

在本例中，getDomains()返回一個域，該域可以位于兩個不同的桶中。首先搜索私有域，如果成功完成，即使沒有結果，也會搜索共享域。

public Mono<String> getDomainId(GetDomainIdRequest request) {
  return getPrivateDomainId(request.getName())
    .switchIfEmpty(getSharedDomainId(request.getName()))
    .switchIfEmpty(ExceptionUtils.illegalState(
      "Domain %s not found", request.getName()));
}

也可以用無條目表示一種錯誤條件。在這個示例中，如果沒有找到私有或共享域，就會生成一個新的IllegalStateException并傳遞給使用者。

然而有時，你希望根據無錯誤或空來做決策，但不是根據值本身。雖然可以使用操作符來實現這個邏輯，但人們常常發現，其復雜度要遠遠高于其價值。在本例中，你應該只使用命令式條件語句。

public Mono<String> getDomainId(String domain, String organizationId) {
  return Mono.just(domain)
    .filter(d -> d == null)
    .then(getSharedDomainIds()
      .switchIfEmpty(getPrivateDomainIds(organizationId))
      .next()  // select first returned
      .switchIfEmpty(ExceptionUtils.illegalState("Domain not found")))
    .switchIfEmpty(getPrivateDomainId(domain, organizationId)
      .switchIfEmpty(getSharedDomainId(domain))
      .switchIfEmpty(
          ExceptionUtils.illegalState("Domain %s not found", domain)));
}

這個示例返回給定的組織(一個分級容器)中給定域名的id。這里有兩個分支：如果域為空，則返回組織范圍內第一個共享域或私有域的id。如果域不為空，則搜索顯式的域名，并返回它的id。如果你覺得這段代碼令人迷惑難懂，不要絕望，我們也一樣!

public Mono<String> getDomainId(String domain, String organizationId) {
  if (domain == null) {
    return getSharedDomainIds()
      .switchIfEmpty(getPrivateDomainIds(organizationId))
      .next()
      .switchIfEmpty(ExceptionUtils.illegalState("Domain not found"));
  } else {
    return getPrivateDomainId(domain, organizationId)
      .switchIfEmpty(getSharedDomainId(domain))
      .switchIfEmpty(
          ExceptionUtils.illegalState("Domain %s not found", domain));
    }
}

這個示例效果一樣，但使用的是命令式條件語句。但更容易理解得多了，你覺得呢？

測試

實際上，大多數有用的流都是異步的。這在測試中是有問題的，因為測試框架往往都是同步的，注冊是通過了還是失敗了，在異步結果返回之前就應該有結果了。為了彌補這一點，你必須阻塞主線程，直到返回結果，然后將這些結果發至斷言的主線程中。

@Test
public void noLatch() {
  Mono.just("alpha")
    .subscribeOn(Schedulers.single())
    .subscribe(s -> assertEquals("bravo", s));
}

這個示例在非主線程上發出一個字符串，出人意料地是，通過了測試。這個測試通過的根本原因，就是當它顯然不應該通過的時候，noLatch方法將會完成執行，而沒有拋出一個AssertionError。

@Test
public void latch() throws InterruptedException {
  CountDownLatch latch = new CountDownLatch(1);
  AtomicReference<String> actual = new AtomicReference<>();

  Mono.just("alpha")
    .subscribeOn(Schedulers.single())
    .subscribe(actual::set, t -> latch.countDown(), latch::countDown);

  latch.await();
  assertEquals("bravo", actual.get());
}

這個例子，它使用一個CountDownLatch來確保latch()方法在流完成之后才返回，雖然不可否認它很笨拙。一旦latch 釋放，主線程中的斷言就會拋出一個AssertionError，導致測試失敗。

如果你看了這些代碼，拒絕以這種方式實現你所有的測試，大家一定會體諒你的，我們保證。幸運的是，Reactor 提供了一個StepVerifier類來輔助測試。

對響應式設計的測試需要的不僅僅是阻塞。你通常需要對多個值和預期錯誤進行斷言，同時確保意外錯誤會導致測試失敗。StepVerifier對每一項都有所考慮。

@Test
public void testMultipleValues() {
  Flux.just("alpha", "bravo")
    .as(StepVerifier::create)
    .expectNext("alpha")
    .expectNext("bravo")
    .expectComplete()
    .verify(Duration.ofSeconds(5));
}

在這個示例中，使用StepVerifier來預期精確發出了alpha和bravo，然后流完成。如果其中一個沒有發出，發出了一個額外的元素，或者產生一個錯誤，測試就會失敗。

@Test
public void shareFails() {
  this.domains
    .share(ShareDomainRequest.builder()
      .domain("test-domain")
      .organization("test-organization")
      .build())
    .as(StepVerifier::create)
    .consumeErrorWith(t -> assertThat(t)
      .isInstanceOf(IllegalArgumentException.class)
      .hasMessage("Private domain test-domain does not exist"))
    .verify(Duration.ofSeconds(5));
}

這個例子使用了一些更高級的StepVerifier特性，并不僅斷言已經發出了一個錯誤信號，而且它還是一個IllegalArgumentException，并且消息匹配預期結果。

CountDownLatches

關于響應式框架的一個關鍵問題是，它們只能協調自己的操作和線程模型。許多響應式編程的執行環境將不僅僅只有一個線程(例如Servlet容器)。在這些環境中，響應式編程天然的異步屬性并不是問題。但是，有一些環境，比如上面的測試示例，那里的進程將在任何單獨的線程之前結束。

public static void main(String[] args) {
  Mono.just("alpha")
    .delaySubscription(Duration.ofSeconds(1))
    .subscribeOn(Schedulers.single())
    .subscribe(System.out::println);
}

就像該測試方法一樣，這個main()方法將在alpha發出之前終止。

public static void main(String[] args) throws InterruptedException {
  CountDownLatch latch = new CountDownLatch(1);

  Mono.just("alpha")
    .delaySubscription(Duration.ofSeconds(1))
    .subscribeOn(Schedulers.single())
    .subscribe(System.out::println, t -> latch.countDown(),
               latch::countDown);

    latch.await();
}

就像在該測試示例中一樣，一個CountDownLatch可以確保主線程在流終止之前不會終止，不管它是在什么線程上執行的。

阻塞流

在可預見的將來，在響應式編程中與阻塞api交互會成為一種常見現象。為了在兩者之間架起橋梁，在等待結果的時候會適當地進行阻塞。但是，當以這種方式連接到阻塞API時，會丟失響應式編程的一些好處，比如有效的資源使用。因此，你將希望盡可能長地保持代碼的響應性，直到最后一刻才阻塞。同樣值得注意的是，這個想法的邏輯總結一下就是，一個響應式的API可以被阻塞，但是一個阻塞的API永遠不能成為響應式。

Mono<User> requestUser(String name) {...}

User getUser(String name) {
  return requestUser(name)
    .block();
}

在這個例子中,.block()用于橋接Mono的結果到必須的返回類型。

Flux<User> requestUsers() {...}

List<User> listUsers() {
  return requestUsers()
    .collectList()
    .block();
}

和前面的例子一樣，.block()用于將結果橋接到必須的返回類型，但在此之前，流必須被收集到一個列表中。

錯誤處理

如前所述，錯誤是流經系統的值。這意味著一直都沒有一個合適的點來捕獲異常。但是，你應該將它們作為流的一部分處理，或者作為訂閱者。.Subscribe()方法有0到3個參數，這些參數允許你處理每個條目，如果錯誤成了就對它進行處理，并對流的完成情況進行處理。

public static void main(String[] args) throws InterruptedException {
  CountDownLatch latch = new CountDownLatch(1);

  Flux.concat(Mono.just("alpha"), Mono.error(new