rx編程

① Reactive（響應式）編程

Reactor 和Rxjava是Reactive Programming範例的一個具體實現，可以概括為：

作為反應式編程方向的第一步，Microsoft在.NET生態系統中創建了Reactive Extensions（Rx）庫。然後RxJava在JVM上實現了響應式編程。隨著時間的推移，通過Reactive Streams工作出現了Java的標准化，這一規范定義了JVM上的反應庫的一組介面和交互規則。它的介面已經在父類Flow下集成到Java 9中。

另外Java 8還引入了Stream，它旨在有效地處理數據流（包括原始類型），這些數據流可以在沒有延遲或很少延遲的情況下訪問。它是基於拉的，只能使用一次，缺少與時間相關的操作，並且可以執行並行計算，但無法指定要使用的線程池。但是它還沒有設計用於處理延遲操作，例如I / O操作。其所不支持的特性就是Reactor或RxJava等Reactive API的用武之地。

Reactor 或 Rxjava等反應性API也提供Java 8 Stream等運算符，但它們更適用於任何流序列（不僅僅是集合），並允許定義一個轉換操作的管道，該管道將應用於通過它的數據，這要歸功於方便的流暢API和使用lambdas。它們旨在處理同步或非同步操作，並允許您緩沖，合並，連接或對數據應用各種轉換。

首先考慮一下，為什麼需要這樣的非同步反應式編程庫？現代應用程序可以支持大量並發用戶，即使現代硬體的功能不斷提高，現代軟體的性能仍然是一個關鍵問題。

人們可以通過兩種方式來提高系統的能力：

通常，Java開發人員使用阻塞代碼編寫程序。這種做法很好，直到出現性能瓶頸，此時需要引入額外的線程。但是，資源利用率的這種擴展會很快引入爭用和並發問題。

更糟糕的是，會導致浪費資源。一旦程序涉及一些延遲（特別是I / O，例如資料庫請求或網路調用），資源就會被浪費，因為線程（或許多線程）現在處於空閑狀態，等待數據。

所以並行化方法不是靈丹妙葯，獲得硬體的全部功能是必要的。

第二種方法，尋求現有資源的更高的使用率，可以解決資源浪費問題。通過編寫非同步，非阻塞代碼，您可以使用相同的底層資源將執行切換到另一個活動任務，然後在非同步處理完成後返回到當前線程進行繼續處理。

但是如何在JVM上生成非同步代碼？ Java提供了兩種非同步編程模型：

但是上面兩種方法都有局限性。首先多個callback難以組合在一起，很快導致代碼難以閱讀以及難以維護（稱為「Callback Hell」）:

考慮下面一個例子：在用戶的UI上展示用戶喜歡的top 5個商品的詳細信息，如果不存在的話則調用推薦服務獲取5個；這個功能的實現需要三個服務支持：一個是獲取用戶喜歡的商品的ID的介面（userService.getFavorites），第二個是獲取商品詳情信息介面（favoriteService.getDetails），第三個是推薦商品與商品詳情的服務（suggestionService.getSuggestions），基於callback模式實現上面功能代碼如下：

如上為了實現該功能，我們寫了很多代碼，使用了大量callback,這些代碼比較晦澀難懂,並且存在代碼重復，下面我們使用Reactor來實現等價的功能：

future相比callback要好一些，但盡管CompletableFuture在Java 8上進行了改進，但它們仍然表現不佳。一起編排多個future是可行但是不容易的，它們不支持延遲計算（比如rxjava中的defer操作）和高級錯誤處理，例如下面例子。考慮另外一個例子：首先我們獲取一個id列表，然後根據id分別獲取對應的name和統計數據，然後組合每個id對應的name和統計數據為一個新的數據，最後輸出所有組合對的值，下面我們使用CompletableFuture來實現這個功能，以便保證整個過程是非同步的，並且每個id對應的處理是並發的：

Reactor本身提供了更多的開箱即用的操作符，使用Reactor來實現上面功能代碼如下:

如上代碼使用reactor方式編寫的代碼相比使用CompletableFuture實現相同功能來說，更簡潔，更通俗易懂。

可組合性，指的是編排多個非同步任務的能力，使用先前任務的結果作為後續任務的輸入或以fork-join方式執行多個任務。

編排任務的能力與代碼的可讀性和可維護性緊密相關。隨著非同步過程層數量和復雜性的增加，能夠編寫和讀取代碼變得越來越困難。正如我們所看到的，callback模型很簡單，但其主要缺點之一是，對於復雜的處理，您需要從回調執行回調，本身嵌套在另一個回調中，依此類推。那個混亂被稱為Callback Hell，正如你可以猜到的（或者從經驗中得知），這樣的代碼很難回歸並推理。

Reactor提供了豐富的組合選項，其中代碼反映了抽象過程的組織，並且所有內容通常都保持在同一級別（嵌套最小化）。

原材料可以經歷各種轉換和其他中間步驟，或者是將中間元素聚集在一起形成較大裝配線的一部分。如果在裝配線中某一點出現堵塞，受影響的工作站可向上游發出信號以限制原材料的向下流動。

雖然Reactive Streams規范根本沒有指定運算符，但Reactor或者rxjava等反應庫的最佳附加值之一是它們提供的豐富的運算符。這些涉及很多方面，從簡單的轉換和過濾到復雜的編排和錯誤處理。

在Reactor中，當您編寫Publisher鏈時，默認情況下數據不會啟動。相反，您可以創建非同步過程的抽象描述（這可以幫助重用和組合）。

上游傳播信號也用於實現背壓，我們在裝配線中將其描述為當工作站比上游工作站處理速度慢時向上游線路發送的反饋信號。

這將推模型轉換為推拉式混合模式，如果上游生產了很多元素，則下游可以從上游拉出n個元素。但是如果元素沒有準備好，就會在上游生產出元素後推數據到下游。

② RxJAVA鏈変粈涔堜紭緙虹偣

RxJava姝ｅ湪Android寮鍙戣呬腑鍙樼殑瓚婃潵瓚婃祦琛屻傚敮涓鐨勯棶棰樺氨鏄涓婃墜涓嶅規槗錛屽挨鍏舵槸澶ч儴鍒嗕漢涔嬪墠閮芥槸浣跨敤鍛戒護寮忕紪紼嬭璦銆備絾鏄涓鏃︿綘寮勬槑鐧戒簡錛屼綘灝變細鍙戠幇RxJava鐪熸槸澶媯掍簡銆傝繖閲屼粎浠呮槸甯鍔╀綘浜嗚ВRxJava錛屾暣涓緋誨垪鍏辨湁鍥涚瘒鏂囩珷錛屽笇鏈涗綘鐪嬪畬榪欏洓綃囨枃絝犱箣鍚庤兘澶熶簡瑙RxJava鑳屽悗鐨勬濇兂錛屽苟涓斿枩嬈涓奟xJava銆傚熀紜RxJava鏈鏍稿績鐨勪袱涓涓滆タ鏄疧bservables錛堣瑙傚療鑰咃紝浜嬩歡婧愶級鍜孲ubscribers錛堣傚療鑰咃級銆侽bservables鍙戝嚭涓緋誨垪浜嬩歡錛孲ubscribers澶勭悊榪欎簺浜嬩歡銆傝繖閲岀殑浜嬩歡鍙浠ユ槸浠諱綍浣犳劅鍏磋叮鐨勪笢瑗匡紙瑙︽懜浜嬩歡錛寃eb鎺ュ彛璋冪敤榪斿洖鐨勬暟鎹銆傘傘傦級涓涓狾bservable鍙浠ュ彂鍑洪浂涓鎴栬呭氫釜浜嬩歡錛岀煡閬撶粨鏉熸垨鑰呭嚭閿欍傛瘡鍙戝嚭涓涓浜嬩歡錛屽氨浼氳皟鐢ㄥ畠鐨凷ubscriber鐨刼nNext鏂規硶錛屾渶鍚庤皟鐢⊿ubscriber.onNext()鎴栬匰ubscriber.onError()緇撴潫銆俁xjava鐨勭湅璧鋒潵寰堟兂璁捐℃ā寮忎腑鐨勮傚療鑰呮ā寮忥紝浣嗘槸鏈変竴鐐規槑鏄句笉鍚岋紝閭ｅ氨鏄濡傛灉涓涓狾bserverble娌℃湁浠諱綍鐨勭殑Subscriber錛岄偅涔堣繖涓狾bservable鏄涓嶄細鍙戝嚭浠諱綍浜嬩歡鐨勩侶elloWorld鍒涘緩涓涓狾bservable瀵硅薄寰堢畝鍗曪紝鐩存帴璋冪敤Observable.create鍗沖彲