環(huán)境：projectreactor2020.0.14

1. 前言

在響應(yīng)式編程中，Project Reactor提供了兩個(gè)核心的概念：Mono和Flux。Mono和Flux都是Reactor中的Publisher，它們可以產(chǎn)生并發(fā)布數(shù)據(jù)，然后可以被訂閱和消費(fèi)。這兩個(gè)概念在WebFlux中有著廣泛的應(yīng)用，幫助我們實(shí)現(xiàn)異步和非阻塞的編程模型。

在這個(gè)主題中，我們將深入探討Mono和Flux的基本使用。我們將了解它們?nèi)绾伪粍?chuàng)建，如何訂閱它們的事件，以及如何處理錯(cuò)誤和完成通知。通過學(xué)習(xí)這些內(nèi)容，你將能夠更好地理解WebFlux的響應(yīng)式編程模型，并能夠在你的項(xiàng)目中有效地使用Mono和Flux。

讓我們開始吧！

2. 環(huán)境依賴

<dependency>
  <groupId>io.projectreactor</groupId>
  <artifactId>reactor-core</artifactId>
</dependency>
<dependencyManagement>
  <dependencies>
    <dependency>
      <groupId>io.projectreactor</groupId>
      <artifactId>reactor-bom</artifactId>
      <version>${reactor.version}</version>
      <type>pom</type>
      <scope>import</scope>
    </dependency>
  </dependencies>
</dependencyManagement>

3. Mono & Flux介紹

Flux

Flux表示了0到N個(gè)元素序列，下圖展示了Flux如何轉(zhuǎn)換元素

Flux

一個(gè)Flux<T>是一個(gè)標(biāo)準(zhǔn)的Publisher<T>，它表示一個(gè)由0到N個(gè)發(fā)射項(xiàng)目組成的異步序列，可選地由一個(gè)完成信號或一個(gè)錯(cuò)誤終止。在響應(yīng)式流規(guī)范中，這三種類型的信號轉(zhuǎn)換為對下游訂閱者的onNext、onComplete和onError方法的調(diào)用。

由于可能信號的范圍很大，F(xiàn)lux是通用的反應(yīng)式類型。請注意，所有事件，甚至是終止事件，都是可選的:只有onComplete事件才能表示一個(gè)空的有限序列，但刪除onComplete事件就會得到一個(gè)無限的空序列(沒什么用處，除了關(guān)于取消的測試)。類似地，無限序列不一定是空的。例如，F(xiàn)lux.interval(Duration)產(chǎn)生一個(gè)無限長的Flux<Long>，并從時(shí)鐘發(fā)出規(guī)則的時(shí)標(biāo)。

Mono

Mono表示了0個(gè)或1個(gè)元素序列，下圖展示了Mono如何轉(zhuǎn)換元素

圖片

Mono

Mono<T>是一個(gè)專門的發(fā)布者<T>，它通過onNext信號發(fā)出最多一個(gè)項(xiàng)目，然后以onComplete信號終止(Mono成功，有或沒有值)，或只發(fā)出一個(gè)onError信號(Mono失敗)。

大多數(shù)Mono實(shí)現(xiàn)都希望在調(diào)用onNext之后立即對其訂閱者調(diào)用onComplete。Mono.never()是一個(gè)異常值:它不會發(fā)出任何信號，這在技術(shù)上并沒有被禁止，但在測試之外并不是特別有用。另一方面，onNext和onError的組合是明確禁止的。

Mono只提供了可用于` Flux `的操作符子集，有些操作符(特別是那些將Mono與另一個(gè)`Publisher`結(jié)合的操作符)會切換到`Flux`。例如，Mono#concatWith(Publisher)返回一個(gè)Flux，而Mono#then(Mono)返回另一個(gè)Mono。

注意，你可以使用Mono來表示只有完成概念的無值異步進(jìn)程(類似于Runnable)。要創(chuàng)建一個(gè)，可以使用一個(gè)空的Mono<Void>。

4. Mono & Flux常用操作

Mono常用操作

• 創(chuàng)建元素

Mono.just(T value)方法：創(chuàng)建一個(gè)包含指定值的Mono對象。

Mono.just(10).subscribe(System.out::println) ;

Mono.empty()方法：創(chuàng)建一個(gè)空的Mono對象，即不包含任何元素。

Mono.justOrEmpty(T value)方法：如果指定值不為null，則創(chuàng)建一個(gè)包含該值的Mono對象；否則創(chuàng)建一個(gè)空的Mono對象。

// 輸出10
Mono.justOrEmpty(10).subscribe(System.out::println) ;
// 如果值為null，沒有任何輸出
Mono.justOrEmpty(null).subscribe(System.out::println) ;

圖片

Mono.fromCallable(Callable<? extends T> supplier)方法：創(chuàng)建一個(gè)Mono對象，該對象包含通過調(diào)用給定Callable對象的call()方法得到的返回值。

// 通過Callable方式，我們可以在內(nèi)部執(zhí)行其它一些動作
Mono.fromCallable(() -> 666).subscribe(System.out::println) ;

圖片

Mono.fromSupplier(Supplier<? extends T> supplier)方法：創(chuàng)建一個(gè)Mono對象，該對象包含通過調(diào)用給定Supplier對象的get()方法得到的返回值。

Mono.fromSupplier(() -> 666).subscribe(System.out::println) ;

圖片

Mono.fromFuture(CompletableFuture<? extends Integer> future)方法：創(chuàng)建一個(gè)Mono對象，該對象包含通過調(diào)用給定CompletableFuture對象

Mono.fromFuture(CompletableFuture.supplyAsync(() -> 666)).subscribe(System.out::println) ;

圖片

• 異常處理

下面這個(gè)示例完整的展示了當(dāng)發(fā)生異常后的處理

public static Mono<Users> invoke(Mono<Users> user) {
  return user.flatMap(u -> {
    if ("admin".equals(u.getName())) {
      return Mono.error(new RuntimeException("越權(quán)")) ;
    }
    u.setName(u.getName() + " - ");
    return Mono.just(u) ;
  });
}


public static void main(String[] args) {
  invoke(Mono.just(new Users("admin")))
    .doOnNext(System.out::println)
    .doOnError(e -> {
      System.out.println(e.getMessage()) ;
    })
    // .onErrorResume(e -> Mono.just(new Users(e.getMessage() + " - fallback"))) // 功能更強(qiáng)，可以對捕獲的異常進(jìn)行響應(yīng)的處理，然后再返回一個(gè)值
    .onErrorReturn(new Users("return")) // 捕獲異常，簡單粗暴直接返回一個(gè)靜態(tài)值
    .doOnNext(System.out::println)
    .subscribe(); 


}

• 執(zhí)行結(jié)果

越權(quán)
Users [name=return]

• 錯(cuò)誤操作符

圖片

• 錯(cuò)誤返回操作符

圖片

• 連接操作符

將該Mono的發(fā)射與提供的發(fā)布者連接(不交錯(cuò))。

Mono.just(10).concatWith(Mono.just(20)).subscribe(System.out::println) ;

圖片

• then操作符

該操作符是在當(dāng)前Mono執(zhí)行完成后切換到另外一個(gè)Mono。

Mono.just(10).doOnNext(System.out::println)
  .then(Mono.just(666)) // 切換到另外一個(gè)Mono通道, 忽略之前的Mono元素
  .doOnNext(System.out::println)
  .subscribe();

圖片

Flux常用操作

• 創(chuàng)建元素

just()：直接使用元素創(chuàng)建Flux，即在創(chuàng)建Flux時(shí)拿到數(shù)據(jù)，之后有誰訂閱它，就重新發(fā)送數(shù)據(jù)給訂閱者。

Flux.just(1, 2, 3...)

圖片

fromArray()、fromIterable()和fromStream()：可以從數(shù)組、Iterable對象或Stream對象中創(chuàng)建Flux對象。

Flux.fromArray(new String[]{"1","2","3"});
Flux.fromIterable(List.of("a","b","c"));
Flux.fromStream(List.of("a","b","c").stream());

fromArray

圖片

fromIterable

圖片

fromStream

empty()：創(chuàng)建一個(gè)不包含任何元素，只發(fā)布結(jié)束消息的序列。

圖片

range(int start, int count)：創(chuàng)建包含從start起始的count個(gè)數(shù)量的Integer對象的序列。

Flux.range(1, 10) ;

圖片

• 錯(cuò)誤處理

Flux.error(new RuntimeException("錯(cuò)誤")).onErrorResume(ex -> Mono.just("發(fā)生異常：" + ex.getMessage())).subscribe(System.out::println) ;

error操作符

圖片

onErrorResume操作符

圖片

• when操作符將給定的Publisher<T>聚合成一個(gè)新的Mono，當(dāng)所有給定的來源都完成時(shí)，這個(gè)Mono就會被滿足。錯(cuò)誤將導(dǎo)致未執(zhí)行的結(jié)果被取消，并立即向返回的Mono發(fā)送錯(cuò)誤。

Flux.just(1, 3, 6).flatMap(id -> {
  Mono<Integer> query = Mono.fromSupplier(() -> {
    System.out.println("查詢數(shù)據(jù)...") ;
    return id * 10 ;
  }).delayElement(Duration.ofSeconds(2)) ;
  Mono<String> save = Mono.fromSupplier(() -> {
    System.out.println("保存數(shù)據(jù)...") ;
    return "success - " + id ;
  }) ;
  return Mono.when(query, save) ;
}).doOnComplete(() -> {
  System.out.println("執(zhí)行完成...") ;
}).subscribe() ;

圖片

• filterWhen操作符

Flux.just(1,2,3,4,5,6).concatMap(item -> Mono.just(item).filterWhen(r -> {
  return Mono.just(r % 2 == 0) ;
})).subscribe(System.out::println) ;

圖片

總之，Reactor中的Flux和Mono是響應(yīng)式編程的核心組件，它們提供了豐富的操作符和方法來處理異步數(shù)據(jù)流。因此，對于使用WebFlux的開發(fā)者來說，掌握Reactor的使用是非常重要的。

完畢?。?！