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 或 one rror 信號，

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 發出 one rror
信號，詳細說明無法發出的原因，在正常情況下，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 Reference Guide
Reactor 3 中文指南

轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/160507.html

標籤：Java

上一篇：程式員居然有這么多需要看的書？！！為你精選各大技術領域100+本技術好書和思維導圖，建議收藏！！！

下一篇：java中的一些屬性(2)

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