Reactor Reactor 項目的主要 artifact 是 reactor-core，這是一個基於 Java 8 的實現了響應式流規範的響應式庫。 Reactor 提供了實現 Publisher 的響應式類 Flux 和 Mono，以及豐富的操作符。一個 Flux 代表 0...N 個元素的響 ...

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); // ⑥

序號① 訂閱並觸發響應式流。
序號② 對每個生成的元素進行消費。
序號③ 對正常元素進行消費，對錯誤進行響應處理。
序號④ 對正常元素和錯誤均有響應，還定義了響應流正常完成後的回調。
序號⑤ 對正常元素、錯誤信號和完成信號均有響應，同時也定義了對該 subscribe 返回的 Subscription 的回調處理。
序號⑥ 通過自定義實現 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 測試類。

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