學習響應式編程 Reactor (3) - reactor 基礎

Reactor

Reactor 項目的主要 artifact 是 reactor-core，這是一個基於 Java 8 的實現了響應式流規範的響應式庫。

Reactor 提供了實現 Publisher 的響應式類 Flux 和 Mono，以及豐富的操做符。一個 Flux 表明 0...N 個元素的響應式流；一個 Mono 表明 0|1 個元素的響應式流。

Flux 和 Mono 之間能夠轉換，好比 Flux 的 count 操做（計算流中元素個數）返回 Mono，Mono 的 concatWith 操做（鏈接另外一個響應式流）返回 Flux。

Flux

Flux<T> 是一個可以發出 0 到 N 個元素的標準 Publisher<T>，它會被一個完成（completion）或錯誤（error）信號終止。所以，一個 Flux 的可能結果是 value、completion
或 value，這三個分別會傳遞給訂閱者中的 onNext、onComplete、onError 方法。

注意：全部的信號事件，包括表明終止的信號事件都是可選的。若是沒有 onNext 事件，可是有 onComplete 事件，那麼發出的就是空的有限流；若是去掉 onComplete
就獲得一個 無限的空數據流。無限的數據流能夠不是空的，好比 Flux.interval(Duration) 生成的是一個 Flux ，這是一個無限週期性發出規律整數的時鐘數據流。

下圖展現的是 Flux 基於時間線的彈珠交互圖，經過操做符轉換 Flux 中元素：

• 上面那條線表示的是 Flux 數據流時間線，時間從左至右
• 上面那條線中的彈珠表明示的是 Flux 發出的 數據元素
• 上面那條線最後的垂直線表示的是 Flux 已經完成成功事件
• 中間的箭頭虛線和框表示的是 Flux 中的元素正在被轉換，框內的文字表示的是轉換的方式（包含操做符）
• 下面那條線表示的是 FLux 通過轉換後的新數據流
• 若是因爲某種緣由致使 Flux 的轉換終止，將使用 X 來代替 垂直線

後續在學習操做符的過程當中，咱們將見到不少相似的彈珠圖，請你們詳細瞭解清楚該圖各部分的含義。

Mono

Mono<T> 是一種特殊的Publisher<T>，它最多隻能發出一個元素，而後（可選的）終止於 onComplete 或 onError 信號。

Mono 中的操做符是 Flux 中操做符的子集，即 Flux 中只有部分操做符適用於 Mono，有些操做符是將 Mono 和另外一個 Publisher 鏈接轉換爲 Flux。例如，Mono#concatWith(Publisher
) 轉換爲 Flux，Mono#then(Mono) 返回另外一個 Mono。

注意：可使用 Mono 來建立一個只有完成概念的空值異步處理過程（相似於 Runnable）。

下圖展現的是 Mono 基於時間線的彈珠交互圖：

建立 Flux 和 Mono

如同建立 Java Stream 同樣，Reactor 也爲咱們提供了 多個工廠方法用來建立 Flux 和 Mono，有了 Stream 的基礎，建立的基本方法咱們來快速過一下。

下面的建立方法，若是是 Flux 或 Mono 獨有的，會在方法名前增長類名前綴。

下面的示例代碼中都有用到 subscribe 方法，下面會講到，你們先了解它是響應式流的訂閱方法，用於觸發流，相似於 Java Stream 中的終端操做。

just

使用提供的元素髮出數據而後結束的流。

Mono.just("hello, world").subscribe(System.out::println);
Mono.justOrEmpty(str).subscribe(System.out::println);
Mono.justOrEmpty(optional).subscribe(System.out::println);

Flux.just("hello", "world").subscribe(System.out::println);
Flux.just("hello").subscribe(System.out::println);

Flux#fromXxx

Flux 提供了 fromArray（從數組）、fromIterable（從迭代器）、fromStream（從 Java Stream 流） 的方式來建立 Flux。

String[] array = new String[]{"hello", "reactor", "flux"};
List<String> iterable = Arrays.asList("foo", "bar", "foobar");

Flux.fromArray(array).subscribe(System.out::println);
Flux.fromIterable(iterable).subscribe(System.out::println);
Flux.fromStream(Arrays.stream(array)).subscribe(System.out::println);

Flux#range

從 start 開始構建一個 Flux，該 Flux 僅發出一系列遞增計數的整數。 也就是說，在 start（包括）和 start + count（排除）之間發出整數，而後完成。見圖識意：

Flux.range(3, 5).subscribe(System.out::println);

Flux#interval

在全局計時器上建立一個 Flux，該 Flux 在初始延遲後，發出從0開始並以指定的時間間隔遞增的長整數。 若是未及時產生，則會經過溢出 IllegalStateException 發出 onError
信號，詳細說明沒法發出的緣由。 在正常狀況下，Flux 將永遠不會完成。interval 提供了 3 個重載方法，三者的區別主要在因而否延遲發出、以及使用的調度器。

interval 生成的是一個無限數據流。

Flux<Long> interval(Duration period)
Flux<Long> interval(Duration delay, Duration period)
Flux<Long> interval(Duration delay, Duration period, Scheduler timer)

• 第 1 個方法，沒有延遲，按照 period 的週期當即發出，默認使用 Schedulers.parallel() 調度器
• 第 2 個方法，以 delay 延遲，按照 period 的週期發出，默認使用 Schedulers.parallel() 調度器
• 第 3 個方法，以 delay 延遲，按照 period 的週期發出，使用指定的調度器

見圖識意：

Flux.interval(Duration.ofMillis(30), Duration.ofMillis(500)).subscribe(System.out::println);

empty

生成一個空的有限流。見圖識意：

Flux.empty().subscribe(System.out::println, System.out::println, () -> System.out.println("結束"));

never

生成一個空的無限流。見圖識意：

Flux.never().subscribe(System.out::println, System.out::println, () -> System.out.println("結束"));

error

生成一個錯誤流。error 有 3 個重載方法，它們的主要區別是否當即生成錯誤及是否由 Supplier 提供，見圖識意：

Flux.error(new IllegalStateException(), true)
    .log()
    .subscribe(System.out::println, System.err::println);

其它

Flux 和 Mono 還提供了編程式的建立數據流的方法，諸如 create、generate、push、handle 等的方式，這些內容暫時不是咱們的重點，這裏咱們不細展開，感興趣的可看 Api 進行研究下。

訂閱 Flux 和 Mono

在上面建立 Flux 和 Mono 筆記的示例代碼中，咱們已經提到了 subscribe 訂閱，在 subscribe 訂閱中，Flux 和 Mono 支持 Java 8 Lambda 表達式。下面咱們來看看 Reactor
爲咱們提供了哪些訂閱方法。

subscribe(); // ①

subscribe(Consumer<? super T> consumer);  // ②

subscribe(Consumer<? super T> consumer,
          Consumer<? super Throwable> errorConsumer); // ③

subscribe(Consumer<? super T> consumer,
          Consumer<? super Throwable> errorConsumer,
          Runnable completeConsumer); // ④

subscribe(Consumer<? super T> consumer,
          Consumer<? super Throwable> errorConsumer,
          Runnable completeConsumer,
          Consumer<? super Subscription> subscriptionConsumer); // ⑤

subscribe(Subscriber<? super T> actual); // ⑥

1. 序號① 訂閱並觸發響應式流。
2. 序號② 對每一個生成的元素進行消費。
3. 序號③ 對正常元素進行消費，對錯誤進行響應處理。
4. 序號④ 對正常元素和錯誤均有響應，還定義了響應流正常完成後的回調。
5. 序號⑤ 對正常元素、錯誤信號和完成信號均有響應，同時也定義了 對該 subscribe 返回的 Subscription 的回調處理。
6. 序號⑥ 經過自定義實現 Subscriber 接口來訂閱。

注意：序號⑤ 變量傳遞一個 Subscription 的引用，若是再也不須要更多元素時，能夠經過它來取消訂閱。取消訂閱時，源頭會中止生成數據，並清理相關的資源。取消和清理的操做是在 Disposable 接口中定義的。

來看下序號 ⑤ 的 subscribe 的彈珠圖：

Flux.range(1, 4)
        .subscribe(System.out::println,
                error -> System.err.println("發生錯誤：" + error),
                () -> System.out.println("完成"),
                sub -> {
                    System.out.println("已訂閱");
                    // 理解背壓
                    // 嘗試修改下 request 中的值，看看有啥變化
                    sub.request(10);
                });

注意：序號⑥ 的方式支持背壓等操做，不在咱們本次筆記的範疇，咱們仍是先略過，後期在學習。

補充

在上節咱們講解 Reactor 調試部分時，遺漏了記錄數據流的日誌方法，再此作下補充：除了基於 stack trace 的方式調試分析，咱們還可使用 log
操做符，來跟蹤響應式流並記錄日誌。將它添加到操做鏈上以後，它會讀取每個再其上游的 Flux 和 Mono 事件（包括 onNext、onError、onComplete、Subscribe、Cancel 和 Request）。

// 嘗試交換下 take 和 log 的順序，看看有啥變化
Flux.range(1, 10)
        // .log()
        .take(3)
        .log()
        .subscribe();

總結

本篇咱們介紹了 Reactor 的基礎知識：先是瞭解了 Reactor 爲咱們提供的響應式流類 Flux 和 Mono，以後學習瞭如何建立他們和訂閱他們，由於有以前 Stream
的基礎，想來你們對這些知識點都好理解和接受。

今天的內容就學到這裏，咱們下篇開始學習 Reactor 的操做符。

源碼詳見：https://github.com/crystalxmumu/spring-web-flux-study-note 下 02-reactor-core-learning
模塊下 ReactorBasicLearningTest 測試類。

參考

1. Reactor 3 Reference Guide
2. Reactor 3 中文指南