。

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

條件邏輯

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

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()));
}

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

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

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

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

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

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

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

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。這里有兩個分支：如果域為空，則返回組織范圍內(nèi)第一個共享域或私有域的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));
    }
}

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

測試

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

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

這個示例在非主線程上發(fā)出一個字符串，出人意料地是，通過了測試。這個測試通過的根本原因，就是當(dāng)它顯然不應(yīng)該通過的時候，noLatch方法將會完成執(zhí)行，而沒有拋出一個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()方法在流完成之后才返回，雖然不可否認(rèn)它很笨拙。一旦latch 釋放，主線程中的斷言就會拋出一個AssertionError，導(dǎo)致測試失敗。

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

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

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

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

@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特性，并不僅斷言已經(jīng)發(fā)出了一個錯誤信號，而且它還是一個IllegalArgumentException，并且消息匹配預(yù)期結(jié)果。

CountDownLatches

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

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

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

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可以確保主線程在流終止之前不會終止，不管它是在什么線程上執(zhí)行的。

阻塞流

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

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

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

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

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

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

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

錯誤處理

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

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

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