反應式編程（Reactive Programming）這種新的編程范式越來越受到開發人員的歡迎。在 Java 社區中比較流行的是 RxJava 和 RxJava 2。本文要介紹的是另外一個新的反應式編程庫 Reactor。

反應式編程介紹

反應式編程來源于數據流和變化的傳播，意味著由底層的執行模型負責通過數據流來自動傳播變化。比如求值一個簡單的表達式 c=a+b，當 a 或者 b 的值發生變化時，傳統的編程范式需要對 a+b 進行重新計算來得到 c 的值。如果使用反應式編程，當 a 或者 b 的值發生變化時，c 的值會自動更新。反應式編程最早由 .NET 平臺上的 Reactive Extensions (Rx) 庫來實現。后來遷移到 Java 平臺之后就產生了著名的 RxJava 庫，并產生了很多其他編程語言上的對應實現。在這些實現的基礎上產生了后來的反應式流（Reactive Streams）規范。該規范定義了反應式流的相關接口，并將集成到 Java 9 中。

在傳統的編程范式中，我們一般通過迭代器（Iterator）模式來遍歷一個序列。這種遍歷方式是由調用者來控制節奏的，采用的是拉的方式。每次由調用者通過 next()方法來獲取序列中的下一個值。使用反應式流時采用的則是推的方式，即常見的發布者-訂閱者模式。當發布者有新的數據產生時，這些數據會被推送到訂閱者來進行處理。在反應式流上可以添加各種不同的操作來對數據進行處理，形成數據處理鏈。這個以聲明式的方式添加的處理鏈只在訂閱者進行訂閱操作時才會真正執行。

反應式流中第一個重要概念是負壓（backpressure）。在基本的消息推送模式中，當消息發布者產生數據的速度過快時，會使得消息訂閱者的處理速度無法跟上產生的速度，從而給訂閱者造成很大的壓力。當壓力過大時，有可能造成訂閱者本身的奔潰，所產生的級聯效應甚至可能造成整個系統的癱瘓。負壓的作用在于提供一種從訂閱者到生產者的反饋渠道。訂閱者可以通過 request()方法來聲明其一次所能處理的消息數量，而生產者就只會產生相應數量的消息，直到下一次 request()方法調用。這實際上變成了推拉結合的模式。

Reactor 簡介

前面提到的 RxJava 庫是 JVM 上反應式編程的先驅，也是反應式流規范的基礎。RxJava 2 在 RxJava 的基礎上做了很多的更新。不過 RxJava 庫也有其不足的地方。RxJava 產生于反應式流規范之前，雖然可以和反應式流的接口進行轉換，但是由于底層實現的原因，使用起來并不是很直觀。RxJava 2 在設計和實現時考慮到了與規范的整合，不過為了保持與 RxJava 的兼容性，很多地方在使用時也并不直觀。Reactor 則是完全基于反應式流規范設計和實現的庫，沒有 RxJava 那樣的歷史包袱，在使用上更加的直觀易懂。Reactor 也是 Spring 5 中反應式編程的基礎。學習和掌握 Reactor 可以更好地理解 Spring 5 中的相關概念。

在 Java 程序中使用 Reactor 庫非常的簡單，只需要通過 Maven 或 Gradle 來添加對 io.projectreactor:reactor-core 的依賴即可，目前的版本是 3.0.5.RELEASE。

Flux 和 Mono

Flux 和 Mono 是 Reactor 中的兩個基本概念。Flux 表示的是包含 0 到 N 個元素的異步序列。在該序列中可以包含三種不同類型的消息通知：正常的包含元素的消息、序列結束的消息和序列出錯的消息。當消息通知產生時，訂閱者中對應的方法 onNext(), onComplete()和 onError()會被調用。Mono 表示的是包含 0 或者 1 個元素的異步序列。該序列中同樣可以包含與 Flux 相同的三種類型的消息通知。Flux 和 Mono 之間可以進行轉換。對一個 Flux 序列進行計數操作，得到的結果是一個 Mono<Long>對象。把兩個 Mono 序列合并在一起，得到的是一個 Flux 對象。

創建 Flux

有多種不同的方式可以創建 Flux 序列。

Flux 類的靜態方法

第一種方式是通過 Flux 類中的靜態方法。

• just()：可以指定序列中包含的全部元素。創建出來的 Flux 序列在發布這些元素之后會自動結束。
• fromArray()，fromIterable()和 fromStream()：可以從一個數組、Iterable 對象或 Stream 對象中創建 Flux 對象。
• empty()：創建一個不包含任何元素，只發布結束消息的序列。
• error(Throwable error)：創建一個只包含錯誤消息的序列。
• never()：創建一個不包含任何消息通知的序列。
• range(int start, int count)：創建包含從 start 起始的 count 個數量的 Integer 對象的序列。
• interval(Duration period)和 interval(Duration delay, Duration period)：創建一個包含了從 0 開始遞增的 Long 對象的序列。其中包含的元素按照指定的間隔來發布。除了間隔時間之外，還可以指定起始元素發布之前的延遲時間。
• intervalMillis(long period)和 intervalMillis(long delay, long period)：與 interval()方法的作用相同，只不過該方法通過毫秒數來指定時間間隔和延遲時間。

代碼清單 1 中給出了上述這些方法的使用示例。

清單 1. 通過 Flux 類的靜態方法創建 Flux 序列

1 2 3 4 5 6

Flux.just("Hello", "World").subscribe(System.out::println); Flux.fromArray(new Integer[] {1, 2, 3}).subscribe(System.out::println); Flux.empty().subscribe(System.out::println); Flux.range(1, 10).subscribe(System.out::println); Flux.interval(Duration.of(10, ChronoUnit.SECONDS)).subscribe(System.out::println); Flux.intervalMillis(1000).subscribe(System.out::println);

上面的這些靜態方法適合于簡單的序列生成，當序列的生成需要復雜的邏輯時，則應該使用 generate() 或 create() 方法。

generate()方法

generate()方法通過同步和逐一的方式來產生 Flux 序列。序列的產生是通過調用所提供的 SynchronousSink 對象的 next()，complete()和 error(Throwable)方法來完成的。逐一生成的含義是在具體的生成邏輯中，next()方法只能最多被調用一次。在有些情況下，序列的生成可能是有狀態的，需要用到某些狀態對象。此時可以使用 generate()方法的另外一種形式 generate(Callable<S> stateSupplier, BiFunction<S,SynchronousSink<T>,S> generator)，其中 stateSupplier 用來提供初始的狀態對象。在進行序列生成時，狀態對象會作為 generator 使用的第一個參數傳入，可以在對應的邏輯中對該狀態對象進行修改以供下一次生成時使用。

在代碼清單 2中，第一個序列的生成邏輯中通過 next()方法產生一個簡單的值，然后通過 complete()方法來結束該序列。如果不調用 complete()方法，所產生的是一個無限序列。第二個序列的生成邏輯中的狀態對象是一個 ArrayList 對象。實際產生的值是一個隨機數。產生的隨機數被添加到 ArrayList 中。當產生了 10 個數時，通過 complete()方法來結束序列。

清單 2. 使用 generate()方法生成 Flux 序列

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16

Flux.generate(sink -> { ????sink.next("Hello"); ????sink.complete(); }).subscribe(System.out::println); final Random random = new Random(); Flux.generate(ArrayList::new, (list, sink) -> { ????int value = random.nextInt(100); ????list.add(value); ????sink.next(value); ????if (list.size() == 10) { ????????sink.complete(); ????} ????return list; }).subscribe(System.out::println);

create()方法

create()方法與 generate()方法的不同之處在于所使用的是 FluxSink 對象。FluxSink 支持同步和異步的消息產生，并且可以在一次調用中產生多個元素。在代碼清單 3 中，在一次調用中就產生了全部的 10 個元素。

清單 3. 使用 create()方法生成 Flux 序列

1 2 3 4 5 6

Flux.create(sink -> { ????for (int i = 0; i < 10; i++) { ????????sink.next(i); ????} ????sink.complete(); }).subscribe(System.out::println);

創建 Mono

Mono 的創建方式與之前介紹的 Flux 比較相似。Mono 類中也包含了一些與 Flux 類中相同的靜態方法。這些方法包括 just()，empty()，error()和 never()等。除了這些方法之外，Mono 還有一些獨有的靜態方法。

• fromCallable()、fromCompletionStage()、fromFuture()、fromRunnable()和 fromSupplier()：分別從 Callable、CompletionStage、CompletableFuture、Runnable 和 Supplier 中創建 Mono。
• delay(Duration duration)和 delayMillis(long duration)：創建一個 Mono 序列，在指定的延遲時間之后，產生數字 0 作為唯一值。
• ignoreElements(Publisher<T> source)：創建一個 Mono 序列，忽略作為源的 Publisher 中的所有元素，只產生結束消息。
• justOrEmpty(Optional<? extends T> data)和 justOrEmpty(T data)：從一個 Optional 對象或可能為 null 的對象中創建 Mono。只有 Optional 對象中包含值或對象不為 null 時，Mono 序列才產生對應的元素。

還可以通過 create()方法來使用 MonoSink 來創建 Mono。代碼清單 4 中給出了創建 Mono 序列的示例。

清單 4. 創建 Mono 序列

1 2 3

Mono.fromSupplier(() -> "Hello").subscribe(System.out::println); Mono.justOrEmpty(Optional.of("Hello")).subscribe(System.out::println); Mono.create(sink -> sink.success("Hello")).subscribe(System.out::println);

操作符

和 RxJava 一樣，Reactor 的強大之處在于可以在反應式流上通過聲明式的方式添加多種不同的操作符。下面對其中重要的操作符進行分類介紹。

buffer 和 bufferTimeout

這兩個操作符的作用是把當前流中的元素收集到集合中，并把集合對象作為流中的新元素。在進行收集時可以指定不同的條件：所包含的元素的最大數量或收集的時間間隔。方法 buffer()僅使用一個條件，而 bufferTimeout()可以同時指定兩個條件。指定時間間隔時可以使用 Duration 對象或毫秒數，即使用 bufferMillis()或 bufferTimeoutMillis()兩個方法。

除了元素數量和時間間隔之外，還可以通過 bufferUntil 和 bufferWhile 操作符來進行收集。這兩個操作符的參數是表示每個集合中的元素所要滿足的條件的 Predicate 對象。bufferUntil 會一直收集直到 Predicate 返回為 true。使得 Predicate 返回 true 的那個元素可以選擇添加到當前集合或下一個集合中；bufferWhile 則只有當 Predicate 返回 true 時才會收集。一旦值為 false，會立即開始下一次收集。

代碼清單 5 給出了 buffer 相關操作符的使用示例。第一行語句輸出的是 5 個包含 20 個元素的數組；第二行語句輸出的是 2 個包含了 10 個元素的數組；第三行語句輸出的是 5 個包含 2 個元素的數組。每當遇到一個偶數就會結束當前的收集；第四行語句輸出的是 5 個包含 1 個元素的數組，數組里面包含的只有偶數。

需要注意的是，在代碼清單 5 中，首先通過 toStream()方法把 Flux 序列轉換成 Java 8 中的 Stream 對象，再通過 forEach()方法來進行輸出。這是因為序列的生成是異步的，而轉換成 Stream 對象可以保證主線程在序列生成完成之前不會退出，從而可以正確地輸出序列中的所有元素。

清單 5. buffer 相關操作符的使用示例

1 2 3 4

Flux.range(1, 100).buffer(20).subscribe(System.out::println); Flux.intervalMillis(100).bufferMillis(1001).take(2).toStream().forEach(System.out::println); Flux.range(1, 10).bufferUntil(i -> i % 2 == 0).subscribe(System.out::println); Flux.range(1, 10).bufferWhile(i -> i % 2 == 0).subscribe(System.out::println);

filter

對流中包含的元素進行過濾，只留下滿足 Predicate 指定條件的元素。代碼清單 6 中的語句輸出的是 1 到 10 中的所有偶數。

清單 6. filter 操作符使用示例

1	Flux.range(1, 10).filter(i -> i % 2 == 0).subscribe(System.out::println);

window

window 操作符的作用類似于 buffer，所不同的是 window 操作符是把當前流中的元素收集到另外的 Flux 序列中，因此返回值類型是 Flux<Flux<T>>。在代碼清單 7 中，兩行語句的輸出結果分別是 5 個和 2 個 UnicastProcessor 字符。這是因為 window 操作符所產生的流中包含的是 UnicastProcessor 類的對象，而 UnicastProcessor 類的 toString 方法輸出的就是 UnicastProcessor 字符。

清單 7. window 操作符使用示例

1 2	Flux.range(1, 100).window(20).subscribe(System.out::println); Flux.intervalMillis(100).windowMillis(1001).take(2).toStream().forEach(System.out::println);

zipWith

zipWith 操作符把當前流中的元素與另外一個流中的元素按照一對一的方式進行合并。在合并時可以不做任何處理，由此得到的是一個元素類型為 Tuple2 的流；也可以通過一個 BiFunction 函數對合并的元素進行處理，所得到的流的元素類型為該函數的返回值。

在代碼清單 8 中，兩個流中包含的元素分別是 a，b 和 c，d。第一個 zipWith 操作符沒有使用合并函數，因此結果流中的元素類型為 Tuple2；第二個 zipWith 操作通過合并函數把元素類型變為 String。

清單 8. zipWith 操作符使用示例

1 2 3 4 5 6

Flux.just("a", "b") ????????.zipWith(Flux.just("c", "d")) ????????.subscribe(System.out::println); Flux.just("a", "b") ????????.zipWith(Flux.just("c", "d"), (s1, s2) -> String.format("%s-%s", s1, s2)) ????????.subscribe(System.out::println);

take

take 系列操作符用來從當前流中提取元素。提取的方式可以有很多種。

• take(long n)，take(Duration timespan)和 takeMillis(long timespan)：按照指定的數量或時間間隔來提取。
• takeLast(long n)：提取流中的最后 N 個元素。
• takeUntil(Predicate<? super T> predicate)：提取元素直到 Predicate 返回 true。
• takeWhile(Predicate<? super T> continuePredicate)： 當 Predicate 返回 true 時才進行提取。
• takeUntilOther(Publisher<?> other)：提取元素直到另外一個流開始產生元素。

在代碼清單 9 中，第一行語句輸出的是數字 1 到 10；第二行語句輸出的是數字 991 到 1000；第三行語句輸出的是數字 1 到 9；第四行語句輸出的是數字 1 到 10，使得 Predicate 返回 true 的元素也是包含在內的。

清單 9. take 系列操作符使用示例

1 2 3 4

Flux.range(1, 1000).take(10).subscribe(System.out::println); Flux.range(1, 1000).takeLast(10).subscribe(System.out::println); Flux.range(1, 1000).takeWhile(i -> i <?10).subscribe(System.out::println); Flux.range(1, 1000).takeUntil(i -> i == 10).subscribe(System.out::println);

reduce 和 reduceWith

reduce 和 reduceWith 操作符對流中包含的所有元素進行累積操作，得到一個包含計算結果的 Mono 序列。累積操作是通過一個 BiFunction 來表示的。在操作時可以指定一個初始值。如果沒有初始值，則序列的第一個元素作為初始值。

在代碼清單 10 中，第一行語句對流中的元素進行相加操作，結果為 5050；第二行語句同樣也是進行相加操作，不過通過一個 Supplier 給出了初始值為 100，所以結果為 5150。

清單 10. reduce 和 reduceWith 操作符使用示例

1 2	Flux.range(1, 100).reduce((x, y) -> x + y).subscribe(System.out::println); Flux.range(1, 100).reduceWith(() -> 100, (x, y) -> x + y).subscribe(System.out::println);

merge 和 mergeSequential

merge 和 mergeSequential 操作符用來把多個流合并成一個 Flux 序列。不同之處在于 merge 按照所有流中元素的實際產生順序來合并，而 mergeSequential 則按照所有流被訂閱的順序，以流為單位進行合并。

代碼清單 11 中分別使用了 merge 和 mergeSequential 操作符。進行合并的流都是每隔 100 毫秒產生一個元素，不過第二個流中的每個元素的產生都比第一個流要延遲 50 毫秒。在使用 merge 的結果流中，來自兩個流的元素是按照時間順序交織在一起；而使用 mergeSequential 的結果流則是首先產生第一個流中的全部元素，再產生第二個流中的全部元素。

清單 11. merge 和 mergeSequential 操作符使用示例

1 2 3 4 5 6

Flux.merge(Flux.intervalMillis(0, 100).take(5), Flux.intervalMillis(50, 100).take(5)) ????????.toStream() ????????.forEach(System.out::println); Flux.mergeSequential(Flux.intervalMillis(0, 100).take(5), Flux.intervalMillis(50, 100).take(5)) ????????.toStream() ????????.forEach(System.out::println);

flatMap 和 flatMapSequential

flatMap 和 flatMapSequential 操作符把流中的每個元素轉換成一個流，再把所有流中的元素進行合并。flatMapSequential 和 flatMap 之間的區別與 mergeSequential 和 merge 之間的區別是一樣的。

在代碼清單 12 中，流中的元素被轉換成每隔 100 毫秒產生的數量不同的流，再進行合并。由于第一個流中包含的元素數量較少，所以在結果流中一開始是兩個流的元素交織在一起，然后就只有第二個流中的元素。

清單 12. flatMap 操作符使用示例

1 2 3 4

Flux.just(5, 10) ????????.flatMap(x -> Flux.intervalMillis(x * 10, 100).take(x)) ????????.toStream() ????????.forEach(System.out::println);

concatMap

concatMap 操作符的作用也是把流中的每個元素轉換成一個流，再把所有流進行合并。與 flatMap 不同的是，concatMap 會根據原始流中的元素順序依次把轉換之后的流進行合并；與 flatMapSequential 不同的是，concatMap 對轉換之后的流的訂閱是動態進行的，而 flatMapSequential 在合并之前就已經訂閱了所有的流。

代碼清單 13 與代碼清單 12 類似，只不過把 flatMap 換成了 concatMap，結果流中依次包含了第一個流和第二個流中的全部元素。

清單 13. concatMap 操作符使用示例

1 2 3 4

Flux.just(5, 10) ????????.concatMap(x -> Flux.intervalMillis(x * 10, 100).take(x)) ????????.toStream() ????????.forEach(System.out::println);

combineLatest

combineLatest 操作符把所有流中的最新產生的元素合并成一個新的元素，作為返回結果流中的元素。只要其中任何一個流中產生了新的元素，合并操作就會被執行一次，結果流中就會產生新的元素。在 代碼清單 14 中，流中最新產生的元素會被收集到一個數組中，通過 Arrays.toString 方法來把數組轉換成 String。

清單 14. combineLatest 操作符使用示例

1 2 3 4 5

Flux.combineLatest( ????????Arrays::toString, ????????Flux.intervalMillis(100).take(5), ????????Flux.intervalMillis(50, 100).take(5) ).toStream().forEach(System.out::println);

消息處理

當需要處理 Flux 或 Mono 中的消息時，如之前的代碼清單所示，可以通過 subscribe 方法來添加相應的訂閱邏輯。在調用 subscribe 方法時可以指定需要處理的消息類型。可以只處理其中包含的正常消息，也可以同時處理錯誤消息和完成消息。代碼清單 15 中通過 subscribe()方法同時處理了正常消息和錯誤消息。

清單 15. 通過 subscribe()方法處理正常和錯誤消息

1 2 3

Flux.just(1, 2) ????????.concatWith(Mono.error(new IllegalStateException())) ????????.subscribe(System.out::println, System.err::println);

正常的消息處理相對簡單。當出現錯誤時，有多種不同的處理策略。第一種策略是通過 onErrorReturn()方法返回一個默認值。在代碼清單 16 中，當出現錯誤時，流會產生默認值 0.

清單 16. 出現錯誤時返回默認值

1 2 3 4

Flux.just(1, 2) ????????.concatWith(Mono.error(new IllegalStateException())) ????????.onErrorReturn(0) ????????.subscribe(System.out::println);

第二種策略是通過 switchOnError()方法來使用另外的流來產生元素。在代碼清單 17 中，當出現錯誤時，將產生 Mono.just(0)對應的流，也就是數字 0。

清單 17. 出現錯誤時使用另外的流

1 2 3 4

Flux.just(1, 2) ????????.concatWith(Mono.error(new IllegalStateException())) ????????.switchOnError(Mono.just(0)) ????????.subscribe(System.out::println);

第三種策略是通過 onErrorResumeWith()方法來根據不同的異常類型來選擇要使用的產生元素的流。在代碼清單 18 中，根據異常類型來返回不同的流作為出現錯誤時的數據來源。因為異常的類型為 IllegalArgumentException，所產生的元素為-1。

清單 18. 出現錯誤時根據異常類型來選擇流

1 2 3 4 5 6 7 8 9 10 11

Flux.just(1, 2) ????????.concatWith(Mono.error(new IllegalArgumentException())) ????????.onErrorResumeWith(e -> { ????????????if (e instanceof IllegalStateException) { ????????????????return Mono.just(0); ????????????} else if (e instanceof IllegalArgumentException) { ????????????????return Mono.just(-1); ????????????} ????????????return Mono.empty(); ????????}) ????????.subscribe(System.out::println);

當出現錯誤時，還可以通過 retry 操作符來進行重試。重試的動作是通過重新訂閱序列來實現的。在使用 retry 操作符時可以指定重試的次數。代碼清單 19 中指定了重試次數為 1，所輸出的結果是 1，2，1，2 和錯誤信息。

清單 19. 使用 retry 操作符進行重試

1 2 3 4

Flux.just(1, 2) ????????.concatWith(Mono.error(new IllegalStateException())) ????????.retry(1) ????????.subscribe(System.out::println);

調度器

前面介紹了反應式流和在其上可以進行的各種操作，通過調度器（Scheduler）可以指定這些操作執行的方式和所在的線程。有下面幾種不同的調度器實現。

• 當前線程，通過 Schedulers.immediate()方法來創建。
• 單一的可復用的線程，通過 Schedulers.single()方法來創建。
• 使用彈性的線程池，通過 Schedulers.elastic()方法來創建。線程池中的線程是可以復用的。當所需要時，新的線程會被創建。如果一個線程閑置太長時間，則會被銷毀。該調度器適用于 I/O 操作相關的流的處理。
• 使用對并行操作優化的線程池，通過 Schedulers.parallel()方法來創建。其中的線程數量取決于 CPU 的核的數量。該調度器適用于計算密集型的流的處理。
• 使用支持任務調度的調度器，通過 Schedulers.timer()方法來創建。
• 從已有的 ExecutorService 對象中創建調度器，通過 Schedulers.fromExecutorService()方法來創建。

某些操作符默認就已經使用了特定類型的調度器。比如 intervalMillis()方法創建的流就使用了由 Schedulers.timer()創建的調度器。通過 publishOn()和 subscribeOn()方法可以切換執行操作的調度器。其中 publishOn()方法切換的是操作符的執行方式，而 subscribeOn()方法切換的是產生流中元素時的執行方式。

在代碼清單 20 中，使用 create()方法創建一個新的 Flux 對象，其中包含唯一的元素是當前線程的名稱。接著是兩對 publishOn()和 map()方法，其作用是先切換執行時的調度器，再把當前的線程名稱作為前綴添加。最后通過 subscribeOn()方法來改變流產生時的執行方式。運行之后的結果是[elastic-2] [single-1] parallel-1。最內層的線程名字 parallel-1 來自產生流中元素時使用的 Schedulers.parallel()調度器，中間的線程名稱 single-1 來自第一個 map 操作之前的 Schedulers.single()調度器，最外層的線程名字 elastic-2 來自第二個 map 操作之前的 Schedulers.elastic()調度器。

清單 20. 使用調度器切換操作符執行方式

1 2 3 4 5 6 7 8 9 10 11

Flux.create(sink -> { ????sink.next(Thread.currentThread().getName()); ????sink.complete(); }) .publishOn(Schedulers.single()) .map(x -> String.format("[%s] %s", Thread.currentThread().getName(), x)) .publishOn(Schedulers.elastic()) .map(x -> String.format("[%s] %s", Thread.currentThread().getName(), x)) .subscribeOn(Schedulers.parallel()) .toStream() .forEach(System.out::println);

測試

在對使用 Reactor 的代碼進行測試時，需要用到 io.projectreactor.addons:reactor-test 庫。

使用 StepVerifier

進行測試時的一個典型的場景是對于一個序列，驗證其中所包含的元素是否符合預期。StepVerifier 的作用是可以對序列中包含的元素進行逐一驗證。在代碼清單 21 中，需要驗證的流中包含 a 和 b 兩個元素。通過 StepVerifier.create()方法對一個流進行包裝之后再進行驗證。expectNext()方法用來聲明測試時所期待的流中的下一個元素的值，而 verifyComplete()方法則驗證流是否正常結束。類似的方法還有 verifyError()來驗證流由于錯誤而終止。

清單 21. 使用 StepVerifier 驗證流中的元素

1 2 3 4

StepVerifier.create(Flux.just("a", "b")) ????????.expectNext("a") ????????.expectNext("b") ????????.verifyComplete();

操作測試時間

有些序列的生成是有時間要求的，比如每隔 1 分鐘才產生一個新的元素。在進行測試中，不可能花費實際的時間來等待每個元素的生成。此時需要用到 StepVerifier 提供的虛擬時間功能。通過 StepVerifier.withVirtualTime()方法可以創建出使用虛擬時鐘的 StepVerifier。通過 thenAwait(Duration)方法可以讓虛擬時鐘前進。

在代碼清單 22 中，需要驗證的流中包含兩個產生間隔為一天的元素，并且第一個元素的產生延遲是 4 個小時。在通過 StepVerifier.withVirtualTime()方法包裝流之后，expectNoEvent()方法用來驗證在 4 個小時之內沒有任何消息產生，然后驗證第一個元素 0 產生；接著 thenAwait()方法來讓虛擬時鐘前進一天，然后驗證第二個元素 1 產生；最后驗證流正常結束。

清單 22. 操作測試時間

1 2 3 4 5 6 7

StepVerifier.withVirtualTime(() -> Flux.interval(Duration.ofHours(4), Duration.ofDays(1)).take(2)) ????????.expectSubscription() ????????.expectNoEvent(Duration.ofHours(4)) ????????.expectNext(0L) ????????.thenAwait(Duration.ofDays(1)) ????????.expectNext(1L) ????????.verifyComplete();

使用 TestPublisher

TestPublisher 的作用在于可以控制流中元素的產生，甚至是違反反應流規范的情況。在代碼清單 23 中，通過 create()方法創建一個新的 TestPublisher 對象，然后使用 next()方法來產生元素，使用 complete()方法來結束流。TestPublisher 主要用來測試開發人員自己創建的操作符。

清單 23. 使用 TestPublisher 創建測試所用的流

1 2 3 4 5 6 7 8 9

final TestPublisher<String> testPublisher = TestPublisher.create(); testPublisher.next("a"); testPublisher.next("b"); testPublisher.complete(); StepVerifier.create(testPublisher) ????????.expectNext("a") ????????.expectNext("b") ????????.expectComplete();

調試

由于反應式編程范式與傳統編程范式的差異性，使用 Reactor 編寫的代碼在出現問題時比較難進行調試。為了更好的幫助開發人員進行調試，Reactor 提供了相應的輔助功能。

啟用調試模式

當需要獲取更多與流相關的執行信息時，可以在程序開始的地方添加代碼清單 24 中的代碼來啟用調試模式。在調試模式啟用之后，所有的操作符在執行時都會保存額外的與執行鏈相關的信息。當出現錯誤時，這些信息會被作為異常堆棧信息的一部分輸出。通過這些信息可以分析出具體是在哪個操作符的執行中出現了問題。

清單 24. 啟用調試模式

1	Hooks.onOperator(providedHook -> providedHook.operatorStacktrace());

不過當調試模式啟用之后，記錄這些額外的信息是有代價的。一般只有在出現了錯誤之后，再考慮啟用調試模式。但是當為了找到問題而啟用了調試模式之后，之前的錯誤不一定能很容易重現出來。為了減少可能的開銷，可以限制只對特定類型的操作符啟用調試模式。

使用檢查點

另外一種做法是通過 checkpoint 操作符來對特定的流處理鏈來啟用調試模式。代碼清單 25 中，在 map 操作符之后添加了一個名為 test 的檢查點。當出現錯誤時，檢查點名稱會出現在異常堆棧信息中。對于程序中重要或者復雜的流處理鏈，可以在關鍵的位置上啟用檢查點來幫助定位可能存在的問題。

清單 25. 使用 checkpoint 操作符

1	Flux.just(1, 0).map(x -> 1 / x).checkpoint("test").subscribe(System.out::println);

日志記錄

在開發和調試中的另外一項實用功能是把流相關的事件記錄在日志中。這可以通過添加 log 操作符來實現。在代碼清單 26 中，添加了 log 操作符并指定了日志分類的名稱。

清單 26. 使用 log 操作符記錄事件

1	Flux.range(1, 2).log("Range").subscribe(System.out::println);

在實際的運行時，所產生的輸出如代碼清單 27 所示。

清單 27. log 操作符所產生的日志

1 2 3 4 5 6 7 8

13:07:56.735 [main] DEBUG reactor.util.Loggers$LoggerFactory - Using Slf4j logging framework 13:07:56.751 [main] INFO Range - | onSubscribe([Synchronous Fuseable] FluxRange.RangeSubscription) 13:07:56.753 [main] INFO Range - | request(unbounded) 13:07:56.754 [main] INFO Range - | onNext(1) 1 13:07:56.754 [main] INFO Range - | onNext(2) 2 13:07:56.754 [main] INFO Range - | onComplete()

“冷”與“熱”序列

之前的代碼清單中所創建的都是冷序列。冷序列的含義是不論訂閱者在何時訂閱該序列，總是能收到序列中產生的全部消息。而與之對應的熱序列，則是在持續不斷地產生消息，訂閱者只能獲取到在其訂閱之后產生的消息。

在代碼清單 28 中，原始的序列中包含 10 個間隔為 1 秒的元素。通過 publish()方法把一個 Flux 對象轉換成 ConnectableFlux 對象。方法 autoConnect()的作用是當 ConnectableFlux 對象有一個訂閱者時就開始產生消息。代碼 source.subscribe()的作用是訂閱該 ConnectableFlux 對象，讓其開始產生數據。接著當前線程睡眠 5 秒鐘，第二個訂閱者此時只能獲得到該序列中的后 5 個元素，因此所輸出的是數字 5 到 9。

清單 28. 熱序列

1 2 3 4 5 6 7 8 9

final Flux<Long> source = Flux.intervalMillis(1000) ????????.take(10) ????????.publish() ????????.autoConnect(); source.subscribe(); Thread.sleep(5000); source ????????.toStream() ????????.forEach(System.out::println);

小結

反應式編程范式對于習慣了傳統編程范式的開發人員來說，既是一個需要進行思維方式轉變的挑戰，也是一個充滿了更多可能的機會。Reactor 作為一個基于反應式流規范的新的 Java 庫，可以作為反應式應用的基礎。本文對 Reactor 庫做了詳細的介紹，包括 Flux 和 Mono 序列的創建、常用操作符的使用、調度器、錯誤處理以及測試和調試技巧等。

參考資源 (resources)

• 參考 Reactor 的官方網站，了解 Reactor 的更多內容。
• 查看 Reactor 的用戶指南。
• 查看 InfoQ 上的Reactor by Example。
• 查看反應式流規范。

資源下載（Reactor例子）

Reactor教程Mono和Flux例子_monoreactor,flux和mono-Java代碼類資源-CSDN下載

來源：使用Reactor進行反應式編程最全教程_hhaip-CSDN博客_reactor教程

與50位技術專家面對面20年技術見證，附贈技術全景圖

總結

以上是生活随笔為你收集整理的使用Reactor进行反应式编程最全教程的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。