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 測試類。