Reactor詳解

1、快速上手

介紹

2、響應式編程

2.1. 阻塞是對資源的浪費

2.2. 異步可以解決問題嗎？

2.3.1. 可編排性與可讀性

2.3.2. 就像裝配流水線

2.3.3. 操作符（Operators）

2.3.4. subscribe() 之前什么都不會發生

2.3.5. 背壓

2.3.6. 熱（Hot） vs 冷（Cold）

3、核心特性

1、Mono和Flux

2、subscribe()

3、流的取消

Disposable

4、BaseSubscriber

5、背壓（Backpressure ）和請求重塑（Reshape Requests）

1、buffer：緩沖

2、limit：限流

6、以編程方式創建序列-Sink

1、同步環境-generate

2、多線程-create

7、 handle()

8、自定義線程調度

9、錯誤處理

1. Catch and return a static default value. 捕獲異常返回一個靜態默認值

2. Catch and execute an alternative path with a fallback method.

3. Catch and dynamically compute a fallback value. 捕獲并動態計算一個返回值

4. Catch, wrap to a BusinessException, and re-throw.

5. Catch, log an error-specific message, and re-throw.

6. Use the finally block to clean up resources or a Java 7 “try-with-resource” construct.

7. 忽略當前異常，僅通知記錄，繼續推進

10：常用操作

1、快速上手

介紹

Reactor 是一個用于JVM的完全非阻塞的響應式編程框架，具備高效的需求管理（即對 “背壓（backpressure）”的控制）能力。它與 Java 8 函數式 API 直接集成，比如 CompletableFuture， Stream，以及 Duration。它提供了異步序列 API Flux（用于[N]個元素）和 Mono（用于 [0|1]個元素），并完全遵循和實現了“響應式擴展規范”（Reactive Extensions Specification）。

Reactor 的 reactor-ipc 組件還支持非阻塞的進程間通信（inter-process communication, IPC）。 Reactor IPC 為 HTTP（包括 Websockets）、TCP 和 UDP 提供了支持背壓的網絡引擎，從而適合應用于微服務架構。并且完整支持響應式編解碼（reactive encoding and decoding）。

<dependencyManagement> <dependencies><dependency><groupId>io.projectreactor</groupId><artifactId>reactor-bom</artifactId><version>2023.0.0</version><type>pom</type><scope>import</scope></dependency></dependencies>
</dependencyManagement>

<dependencies><dependency><groupId>io.projectreactor</groupId><artifactId>reactor-core</artifactId> </dependency><dependency><groupId>io.projectreactor</groupId><artifactId>reactor-test</artifactId> <scope>test</scope></dependency>
</dependencies>

2、響應式編程

響應式編程是一種關注于數據流（data streams）和變化傳遞（propagation of change）的異步編程方式。這意味著它可以用既有的編程語言表達靜態（如數組）或動態（如事件源）的數據流。

了解歷史：

在響應式編程方面，微軟跨出了第一步，它在 .NET 生態中創建了響應式擴展庫（Reactive Extensions library, Rx）。接著 RxJava 在JVM上實現了響應式編程。后來，在 JVM 平臺出現了一套標準的響應式編程規范，它定義了一系列標準接口和交互規范。并整合到 Java 9 中（使用 Flow 類）。
響應式編程通常作為面向對象編程中的“觀察者模式”（Observer design pattern）的一種擴展。響應式流（reactive streams）與“迭代子模式”（Iterator design pattern）也有相通之處，因為其中也有 Iterable-Iterator 這樣的對應關系。主要的區別在于，Iterator 是基于 “拉取”（pull）方式的，而響應式流是基于“推送”（push）方式的。
使用 iterator 是一種“命令式”（imperative）編程范式，即使訪問元素的方法是 Iterable 的唯一職責。關鍵在于，什么時候執行 next() 獲取元素取決于開發者。在響應式流中，相對應的角色是 Publisher-Subscriber，但是 當有新的值到來的時候 ，卻反過來由發布者（Publisher）通知訂閱者（Subscriber），這種“推送”模式是響應式的關鍵。此外，對推送來的數據的操作是通過一種聲明式（declaratively）而不是命令式（imperatively）的方式表達的：開發者通過描述“控制流程”來定義對數據流的處理邏輯。
除了數據推送，對錯誤處理（error handling）和完成（completion）信號的定義也很完善。一個 Publisher 可以推送新的值到它的 Subscriber（調用 onNext 方法），同樣也可以推送錯誤（調用 onError 方法）和完成（調用 onComplete 方法）信號。錯誤和完成信號都可以終止響應式流。可以用下邊的表達式描述：
```
onNext x 0..N [onError | onComplete]
```

2.1. 阻塞是對資源的浪費

現代應用需要應對大量的并發用戶，而且即使現代硬件的處理能力飛速發展，軟件性能仍然是關鍵因素。

廣義來說我們有兩種思路來提升程序性能：

并行化（parallelize） ：使用更多的線程和硬件資源。[異步]
基于現有的資源來 提高執行效率 。

通常，Java開發者使用阻塞式（blocking）編寫代碼。這沒有問題，在出現性能瓶頸后，我們可以增加處理線程，線程中同樣是阻塞的代碼。但是這種使用資源的方式會迅速面臨資源競爭和并發問題。

更糟糕的是，阻塞會浪費資源。具體來說，比如當一個程序面臨延遲（通常是I/O方面，比如數據庫讀寫請求或網絡調用），所在線程需要進入 idle 狀態等待數據，從而浪費資源。

所以，并行化方式并非銀彈。這是挖掘硬件潛力的方式，但是卻帶來了復雜性，而且容易造成浪費。

2.2. 異步可以解決問題嗎？

第二種思路——提高執行效率——可以解決資源浪費問題。通過編寫 異步非阻塞 的代碼，（任務發起異步調用后）執行過程會切換到另一個 使用同樣底層資源 的活躍任務，然后等異步調用返回結果再去處理。

但是在 JVM 上如何編寫異步代碼呢？Java 提供了兩種異步編程方式：

回調（Callbacks） ：異步方法沒有返回值，而是采用一個 callback 作為參數（lambda 或匿名類），當結果出來后回調這個 callback。常見的例子比如 Swings 的 EventListener。
Futures ：異步方法立即返回一個 Future<T>，該異步方法要返回結果的是 T 類型，通過 Future封裝。這個結果并不是立刻可以拿到，而是等實際處理結束才可用。比如， ExecutorService 執行 Callable<T> 任務時會返回 Future 對象。

這些技術夠用嗎？并非對于每個用例都是如此，兩種方式都有局限性。

回調很難組合起來，因為很快就會導致代碼難以理解和維護（即所謂的“回調地獄（callback hell）”）。

考慮這樣一種情景：

在用戶界面上顯示用戶的5個收藏，或者如果沒有任何收藏提供5個建議。
這需要3個服務（一個提供收藏的ID列表，第二個服務獲取收藏內容，第三個提供建議內容）：

回調地獄（Callback Hell）的例子：

userService.getFavorites(userId, new Callback<List<String>>() { public void onSuccess(List<String> list) { if (list.isEmpty()) { suggestionService.getSuggestions(new Callback<List<Favorite>>() {public void onSuccess(List<Favorite> list) { UiUtils.submitOnUiThread(() -> { list.stream().limit(5).forEach(uiList::show); });}public void onError(Throwable error) { UiUtils.errorPopup(error);}});} else {list.stream() .limit(5).forEach(favId -> favoriteService.getDetails(favId, new Callback<Favorite>() {public void onSuccess(Favorite details) {UiUtils.submitOnUiThread(() -> uiList.show(details));}public void onError(Throwable error) {UiUtils.errorPopup(error);}}));}}public void onError(Throwable error) {UiUtils.errorPopup(error);}
});

Reactor改造后為

userService.getFavorites(userId) .flatMap(favoriteService::getDetails) .switchIfEmpty(suggestionService.getSuggestions()) .take(5) .publishOn(UiUtils.uiThreadScheduler()) .subscribe(uiList::show, UiUtils::errorPopup);

如果你想確保“收藏的ID”的數據在800ms內獲得（如果超時，從緩存中獲取）呢？在基于回調的代碼中，會比較復雜。但 Reactor 中就很簡單，在處理鏈中增加一個 timeout 的操作符即可。

userService.getFavorites(userId).timeout(Duration.ofMillis(800)) .onErrorResume(cacheService.cachedFavoritesFor(userId)) .flatMap(favoriteService::getDetails) .switchIfEmpty(suggestionService.getSuggestions()).take(5).publishOn(UiUtils.uiThreadScheduler()).subscribe(uiList::show, UiUtils::errorPopup);

背壓（backpressure） 具體來說即 消費者能夠反向告知生產者生產內容的速度的能力
高層次 （同時也是有高價值的）的抽象，從而達到 并發無關 的效果

2.3.1. 可編排性與可讀性

可編排性，指的是編排多個異步任務的能力。比如我們將前一個任務的結果傳遞給后一個任務作為輸入，或者將多個任務以分解再匯總（fork-join）的形式執行，或者將異步的任務作為離散的組件在系統中進行重用。

這種編排任務的能力與代碼的可讀性和可維護性是緊密相關的。隨著異步處理任務數量和復雜度的提高，編寫和閱讀代碼都變得越來越困難。就像我們剛才看到的，回調模式是簡單的，但是缺點是在復雜的處理邏輯中，回調中會層層嵌入回調，導致 回調地獄（Callback Hell） 。你能猜到（或有過這種痛苦經歷），這樣的代碼是難以閱讀和分析的。

Reactor 提供了豐富的編排操作，從而代碼直觀反映了處理流程，并且所有的操作保持在同一層次（盡量避免了嵌套）。

2.3.2. 就像裝配流水線

你可以想象數據在響應式應用中的處理，就像流過一條裝配流水線。Reactor 既是傳送帶，又是一個個的裝配工或機器人。原材料從源頭（最初的 Publisher）流出，最終被加工為成品，等待被推送到消費者（或者說 Subscriber）。

原材料會經過不同的中間處理過程，或者作為半成品與其他半成品進行組裝。如果某處有齒輪卡住，或者某件產品的包裝過程花費了太久時間，相應的工位就可以向上游發出信號來限制或停止發出原材料。

2.3.3. 操作符（Operators）

在 Reactor 中，操作符（operator）就像裝配線中的工位（操作員或裝配機器人）。每一個操作符對 Publisher 進行相應的處理，然后將 Publisher 包裝為一個新的 Publisher。就像一個鏈條，數據源自第一個 Publisher，然后順鏈條而下，在每個環節進行相應的處理。最終，一個訂閱者 (Subscriber）終結這個過程。請記住，在訂閱者（Subscriber）訂閱（subscribe）到一個發布者（Publisher）之前，什么都不會發生。

理解了操作符會創建新的 Publisher 實例這一點，能夠幫助你避免一個常見的問題，這種問題會讓你覺得處理鏈上的某個操作符沒有起作用。

雖然響應式流規范（Reactive Streams specification）沒有規定任何操作符，類似 Reactor 這樣的響應式庫所帶來的最大附加價值之一就是提供豐富的操作符。包括基礎的轉換操作，到過濾操作，甚至復雜的編排和錯誤處理操作。

2.3.4. subscribe() 之前什么都不會發生

在 Reactor 中，當你創建了一條 Publisher 處理鏈，數據還不會開始生成。事實上，你是創建了一種抽象的對于異步處理流程的描述（從而方便重用和組裝）。

當真正“訂閱（subscrib）”的時候，你需要將 Publisher 關聯到一個 Subscriber 上，然后才會觸發整個鏈的流動。這時候，Subscriber 會向上游發送一個 request 信號，一直到達源頭的 Publisher。

2.3.5. 背壓

向上游傳遞信號這一點也被用于實現背壓，就像在裝配線上，某個工位的處理速度如果慢于流水線速度，會對上游發送反饋信號一樣。

在響應式流規范中實際定義的機制同剛才的類比非常接近：訂閱者可以無限接受數據并讓它的源頭 “滿負荷”推送所有的數據，也可以通過使用 request 機制來告知源頭它一次最多能夠處理 n 個元素。

中間環節的操作也可以影響 request。想象一個能夠將每10個元素分批打包的緩存（buffer）操作。如果訂閱者請求一個元素，那么對于源頭來說可以生成10個元素。此外預取策略也可以使用了，比如在訂閱前預先生成元素。

這樣能夠將“推送”模式轉換為“推送+拉取”混合的模式，如果下游準備好了，可以從上游拉取 n 個元素；但是如果上游元素還沒有準備好，下游還是要等待上游的推送。

2.3.6. 熱（Hot） vs 冷（Cold）

在 Rx 家族的響應式庫中，響應式流分為“熱”和“冷”兩種類型，區別主要在于響應式流如何對訂閱者進行響應：

一個“冷”的序列，指對于每一個 Subscriber，都會收到從頭開始所有的數據。如果源頭生成了一個 HTTP 請求，對于每一個訂閱都會創建一個新的 HTTP 請求。
一個“熱”的序列，指對于一個 Subscriber，只能獲取從它開始訂閱之后發出的數據。不過注意，有些“熱”的響應式流可以緩存部分或全部歷史數據。通常意義上來說，一個“熱”的響應式流，甚至在即使沒有訂閱者接收數據的情況下，也可以發出數據（這一點同 “Subscribe() 之前什么都不會發生”的規則有沖突）。

3、核心特性

1、Mono和Flux

Mono: 0|1 數據流

Flux: N數據流

響應式流：元素（內容） + 信號（完成/異常）；

2、subscribe()

自定義流的信號感知回調

flux.subscribe(v-> System.out.println("v = " + v), //流元素消費throwable -> System.out.println("throwable = " + throwable), //感知異常結束()-> System.out.println("流結束了...") //感知正常結束
);

自定義消費者

flux.subscribe(new BaseSubscriber<String>() {
?// 生命周期鉤子1： 訂閱關系綁定的時候觸發@Overrideprotected void hookOnSubscribe(Subscription subscription) {// 流被訂閱的時候觸發System.out.println("綁定了..."+subscription);
?//找發布者要數據request(1); //要1個數據
// ? ? ? ? ? ? ?  requestUnbounded(); //要無限數據}
?@Overrideprotected void hookOnNext(String value) {System.out.println("數據到達，正在處理："+value);request(1); //要1個數據}
?
?//  hookOnComplete、hookOnError 二選一執行@Overrideprotected void hookOnComplete() {System.out.println("流正常結束...");}
?@Overrideprotected void hookOnError(Throwable throwable) {System.out.println("流異常..."+throwable);}
?@Overrideprotected void hookOnCancel() {System.out.println("流被取消...");}
?@Overrideprotected void hookFinally(SignalType type) {System.out.println("最終回調...一定會被執行");}});

3、流的取消

消費者調用 cancle() 取消流的訂閱；

Disposable

? ?

 ? ?Flux<String> flux = Flux.range(1, 10).map(i -> {System.out.println("map..."+i);if(i==9) {i = 10/(9-i); //數學運算異常;  doOnXxx}return "哈哈：" + i;}); //流錯誤的時候，把錯誤吃掉，轉為正常信號
?
?
// ? ? ?  flux.subscribe(); //流被訂閱; 默認訂閱；
// ? ? ?  flux.subscribe(v-> System.out.println("v = " + v));//指定訂閱規則： 正常消費者：只消費正常元素
?
?
// ? ? ?  flux.subscribe(
// ? ? ? ? ? ? ?  v-> System.out.println("v = " + v), //流元素消費
// ? ? ? ? ? ? ?  throwable -> System.out.println("throwable = " + throwable), //感知異常結束
// ? ? ? ? ? ? ?  ()-> System.out.println("流結束了...") //感知正常結束
// ? ? ?  );
?
?// 流的生命周期鉤子可以傳播給訂閱者。//  a() {// ? ?  data = b();//  }flux.subscribe(new BaseSubscriber<String>() {
?// 生命周期鉤子1： 訂閱關系綁定的時候觸發@Overrideprotected void hookOnSubscribe(Subscription subscription) {// 流被訂閱的時候觸發System.out.println("綁定了..."+subscription);
?//找發布者要數據request(1); //要1個數據
// ? ? ? ? ? ? ?  requestUnbounded(); //要無限數據}
?@Overrideprotected void hookOnNext(String value) {System.out.println("數據到達，正在處理："+value);if(value.equals("哈哈：5")){cancel(); //取消流}request(1); //要1個數據}
?
?//  hookOnComplete、hookOnError 二選一執行@Overrideprotected void hookOnComplete() {System.out.println("流正常結束...");}
?@Overrideprotected void hookOnError(Throwable throwable) {System.out.println("流異常..."+throwable);}
?@Overrideprotected void hookOnCancel() {System.out.println("流被取消...");}
?@Overrideprotected void hookFinally(SignalType type) {System.out.println("最終回調...一定會被執行");}});

4、BaseSubscriber

自定義消費者，推薦直接編寫 BaseSubscriber 的邏輯；

5、背壓（Backpressure ）和請求重塑（Reshape Requests）

1、buffer：緩沖

Flux<List<Integer>> flux = Flux.range(1, 10) ?//原始流10個.buffer(3).log();//緩沖區：緩沖3個元素: 消費一次最多可以拿到三個元素； 湊滿數批量發給消費者
//
// ? ? ?  //一次發一個，一個一個發；
// 10元素，buffer(3)；消費者請求4次，數據消費完成

2、limit：限流

Flux.range(1, 1000).log()//限流觸發，看上游是怎么限流獲取數據的.limitRate(100) //一次預取30個元素； 第一次 request(100)，以后request(75).subscribe();

6、以編程方式創建序列-Sink

Sink.next

Sink.complete

1、同步環境-generate

2、多線程-create

7、 handle()

自定義流中元素處理規則

  ?//Flux.range(1,10).handle((value,sink)->{System.out.println("拿到的值："+value);sink.next("張三："+value); //可以向下發送數據的通道}).log() //日志.subscribe();

8、自定義線程調度

響應式：響應式編程：全異步、消息、事件回調

默認還是用當前線程，生成整個流、發布流、流操作

public void thread1(){Scheduler s = Schedulers.newParallel("parallel-scheduler", 4);
?final Flux<String> flux = Flux.range(1, 2).map(i -> 10 + i).log().publishOn(s).map(i -> "value " + i);
?//只要不指定線程池，默認發布者用的線程就是訂閱者的線程；new Thread(() -> flux.subscribe(System.out::println)).start();
}

9、錯誤處理

命令式編程：常見的錯誤處理方式

1. Catch and return a static default value. 捕獲異常返回一個靜態默認值

try {return doSomethingDangerous(10);
}
catch (Throwable error) {return "RECOVERED";
}

onErrorReturn: 實現上面效果，錯誤的時候返回一個值

1、吃掉異常，消費者無異常感知
2、返回一個兜底默認值

3、流正常完成；

 ? ?Flux.just(1, 2, 0, 4).map(i -> "100 / " + i + " = " + (100 / i)).onErrorReturn(NullPointerException.class,"哈哈-6666").subscribe(v-> System.out.println("v = " + v),err -> System.out.println("err = " + err),()-> System.out.println("流結束")); // error handling example

2. Catch and execute an alternative path with a fallback method.

吃掉異常，執行一個兜底方法；

try {return doSomethingDangerous(10);
}
catch (Throwable error) {return doOtherthing(10);
}

onErrorResume

1、吃掉異常，消費者無異常感知
2、調用一個兜底方法
3、流正常完成

Flux.just(1, 2, 0, 4).map(i -> "100 / " + i + " = " + (100 / i)).onErrorResume(err -> Mono.just("哈哈-777")).subscribe(v -> System.out.println("v = " + v),err -> System.out.println("err = " + err),() -> System.out.println("流結束"));

3. Catch and dynamically compute a fallback value. 捕獲并動態計算一個返回值

根據錯誤返回一個新值

try {Value v = erroringMethod();return MyWrapper.fromValue(v);
}
catch (Throwable error) {return MyWrapper.fromError(error);
}
.onErrorResume(err -> Flux.error(new BusinessException(err.getMessage()+"：炸了")))

4. Catch, wrap to a BusinessException, and re-throw.

捕獲并包裝成一個業務異常，并重新拋出

try {return callExternalService(k);
}
catch (Throwable error) {throw new BusinessException("oops, SLA exceeded", error);
}

包裝重新拋出異常: 推薦用 .onErrorMap

1、吃掉異常，消費者有感知
2、拋新異常
3、流異常完成

.onErrorResume(err -> Flux.error(new BusinessException(err.getMessage()+"：炸了")))Flux.just(1, 2, 0, 4).map(i -> "100 / " + i + " = " + (100 / i)).onErrorMap(err-> new BusinessException(err.getMessage()+": 又炸了...")).subscribe(v -> System.out.println("v = " + v),err -> System.out.println("err = " + err),() -> System.out.println("流結束"));

5. Catch, log an error-specific message, and re-throw.

捕獲異常，記錄特殊的錯誤日志，重新拋出

try {
?return callExternalService(k);
?
}
?
catch (RuntimeException error) {
?//make a record of the error
?log("uh oh, falling back, service failed for key " + k);
?throw error;
?
}Flux.just(1, 2, 0, 4)
?
? ? ? ? ? ? ? ?  .map(i -> "100 / " + i + " = " + (100 / i))
?
? ? ? ? ? ? ? ?  .doOnError(err -> {
?
? ? ? ? ? ? ? ? ? ? ?System.out.println("err已被記錄 = " + err);
?
? ? ? ? ? ? ? ?  }).subscribe(v -> System.out.println("v = " + v),
?
? ? ? ? ? ? ? ? ? ? ? ? ?err -> System.out.println("err = " + err),
?
? ? ? ? ? ? ? ? ? ? ? ?  () -> System.out.println("流結束"));

異常被捕獲、做自己的事情
不影響異常繼續順著流水線傳播
1、不吃掉異常，只在異常發生的時候做一件事，消費者有感知

6. Use the finally block to clean up resources or a Java 7 “try-with-resource” construct.

?Flux.just(1, 2, 3, 4).map(i -> "100 / " + i + " = " + (100 / i)).doOnError(err -> {System.out.println("err已被記錄 = " + err);}).doFinally(signalType -> {System.out.println("流信號："+signalType);})

7. 忽略當前異常，僅通知記錄，繼續推進

Flux.just(1,2,3,0,5).map(i->10/i).onErrorContinue((err,val)->{System.out.println("err = " + err);System.out.println("val = " + val);System.out.println("發現"+val+"有問題了，繼續執行其他的，我會記錄這個問題");}) //發生.subscribe(v-> System.out.println("v = " + v),err-> System.out.println("err = " + err));

10：常用操作

filter、flatMap、concatMap、flatMapMany、transform、defaultIfEmpty、switchIfEmpty、concat、concatWith、merge、mergeWith、mergeSequential、zip、zipWith...

今日內容：

常用操作
錯誤處理
超時與重試
Sinks工具類
- 單播
- 多播
- 重放
- 背壓
- 緩存
阻塞式API
- block
Context-API：響應式中的ThreadLocal

ThreadLocal機制失效

 ? ?Flux.just(1,2,3).transformDeferredContextual((flux,context)->{System.out.println("flux = " + flux);System.out.println("context = " + context);return flux.map(i->i+"==>"+context.get("prefix"));})//上游能拿到下游的最近一次數據.contextWrite(Context.of("prefix","哈哈"))//ThreadLocal共享了數據，上游的所有人能看到; Context由下游傳播給上游.subscribe(v-> System.out.println("v = " + v));

ParallelFlux：
- 并發流

 ? ?Flux.range(1,1000000).buffer(100).parallel(8).runOn(Schedulers.newParallel("yy"))
.log()
.subscribe();

Reactor詳解

1、快速上手

介紹

2、響應式編程

2.1. 阻塞是對資源的浪費

2.2. 異步可以解決問題嗎？

2.3.1. 可編排性與可讀性

2.3.2. 就像裝配流水線

2.3.3. 操作符（Operators）

2.3.4. subscribe() 之前什么都不會發生

2.3.5. 背壓

2.3.6. 熱（Hot） vs 冷（Cold）

3、核心特性

1、Mono和Flux

2、subscribe()

3、流的取消

Disposable

4、BaseSubscriber

5、背壓（Backpressure ）和請求重塑（Reshape Requests）

1、buffer：緩沖

2、limit：限流

6、以編程方式創建序列-Sink

1、同步環境-generate

2、多線程-create

7、 handle()

8、自定義線程調度

9、錯誤處理

1. Catch and return a static default value. 捕獲異常返回一個靜態默認值

2. Catch and execute an alternative path with a fallback method.

3. Catch and dynamically compute a fallback value. 捕獲并動態計算一個返回值

4. Catch, wrap to a BusinessException, and re-throw.

5. Catch, log an error-specific message, and re-throw.

6. Use the finally block to clean up resources or a Java 7 “try-with-resource” construct.

7. 忽略當前異常，僅通知記錄，繼續推進

10：常用操作

相關文章