RxJS是一個強大的Reactive編程庫，提供了強大的數據流組合與控制能力，但是其學習門檻一直很高，本次分享期望從一些特別的角度解讀它在業務中的使用，而不是從API角度去講解。

　　RxJS簡介

　　通常，對RxJS的解釋會是這么一些東西，我們來分別看看它們的含義是什么。

　　Reactive

　　Lodash for events

　　Observable

　　Stream-based

　　什么是Reactive呢，一個比較直觀的對比是這樣的：

　　比如說，abc三個變量之間存在加法關系：

　　a = b + c

　　在傳統方式下，這是一種一次性的賦值過程，調用一次就結束了，后面b和c再改變，a也不會變了。

　　而在Reactive的理念中，我們定義的不是一次性賦值過程，而是可重復的賦值過程，或者說是變量之間的關系：

　　a: = b + c

　　定義出這種關系之后，每次b或者c產生改變，這個表達式都會被重新計算。不同的庫或者語言的實現機制可能不同，寫法也不完全一樣，但理念是相通的，都是描述出數據之間的聯動關系。

　　在前端，我們通常有這么一些方式來處理異步的東西：

　　回調

　　事件

　　Promise

　　Generator

　　其中，存在兩種處理問題的方式，因為需求也是兩種：

　　分發

　　流程

　　在處理分發的需求的時候，回調、事件或者類似訂閱發布這種模式是比較合適的；而在處理流程性質的需求時，Promise和Generator比較合適。

　　在前端，尤其交互很復雜的系統中，RxJS其實是要比Generator有優勢的，因為常見的每種客戶端開發都是基于事件編程的，對于事件的處理會非常多，而一旦系統中大量出現一個事件要修改視圖的多個部分（狀態樹的多個位置），分發關系就更多了。

　　RxJS的優勢在于結合了兩種模式，它的每個Observable上都能夠訂閱，而Observable之間的關系，則能夠體現流程（注意，RxJS里面的流程的控制和處理，其直觀性略強于Promise，但弱于Generator）。

　　我們可以把一切輸入都當做數據流來處理，比如說：

　　用戶操作

　　網絡響應

　　定時器

　　Worker

　　RxJS提供了各種API來創建數據流：

　　單值：of, empty, never

　　多值：from

　　定時：interval, timer

　　從事件創建：fromEvent

　　從Promise創建：fromPromise

　　自定義創建：create

　　創建出來的數據流是一種可觀察的序列，可以被訂閱，也可以被用來做一些轉換操作，比如：

　　改變數據形態：map, mapTo, pluck

　　過濾一些值：filter, skip, first, last, take

　　時間軸上的操作：delay, timeout, throttle, debounce, audit, bufferTime

　　累加：reduce, scan

　　異常處理：throw, catch, retry, finally

　　條件執行：takeUntil, delayWhen, retryWhen, subscribeOn, ObserveOn

　　轉接：switch

　　也可以對若干個數據流進行組合：

　　concat，保持原來的序列順序連接兩個數據流

　　merge，合并序列

　　race，預設條件為其中一個數據流完成

　　forkJoin，預設條件為所有數據流都完成

　　zip，取各來源數據流最后一個值合并為對象

　　combineLatest，取各來源數據流最后一個值合并為數組

　　這時候回頭看，其實RxJS在事件處理的路上已經走得太遠了，從事件到流，它被稱為lodash for events，倒不如說是lodash for stream更貼切，它提供的這些操作符也確實可以跟lodash媲美。

　　數據流這個詞，很多時候，是從data-flow翻譯過來的，但flow跟stream是不一樣的，我的理解是：flow只關注一個大致方向，而stream是受到更嚴格約束的，它更像是在無形的管道里面流動。

　　那么，數據的管道是什么形狀的？

　　在RxJS中，存在這么幾種東西：

　　Observable 可觀察序列，只出不進

　　Observer 觀察者，只進不出

　　Subject 可出可進的可觀察序列，可作為觀察者

　　ReplaySubject 帶回放

　　Subscription 訂閱關系

　　前三種東西，根據它們數據進出的可能性，可以通俗地理解他們的連接方式，這也就是所謂管道的“形狀”，一端密閉一端開頭，還是兩端開口，都可以用來輔助記憶。

　　上面提到的Subscription，則是訂閱之后形成的一個訂閱關系，可以用于取消訂閱。

　　下面，我們通過一些示例來大致了解一下RxJS所提供的能力，以及用它進行開發所需要的思路轉換。

　　示例一：簡單的訂閱

　　很多時候，我們會有一些顯示時間的場景，比如在頁面下添加評論，評論列表中顯示了它們分別是什么時間創建的，為了含義更清晰，可能我們會引入moment這樣的庫，把這個時間轉換為與當前時間的距離：

　　const diff = moment(createAt).fromNow()

　　這樣，顯示的時間就是：一分鐘內，昨天，上個月這樣的字樣。

　　但我們注意到，引入這個轉換是為了增強體驗，而如果某個用戶停留在當前視圖時間太長，它的這些信息會變得不準確，比如說，用戶停留了一個小時，而它看到的信息還顯示：5分鐘之前發表了評論，實際時間是一個小時零5分鐘以前的事了。

　　從這個角度看，我們做這個體驗增強的事情只做了一半，不準確的信息是不能算作增強體驗的。

　　在沒有RxJS的情況下，我們可能會通過一個定時器來做這件事，比如在組件內部：

　　tick() {

　　this.diff = moment(createAt).fromNow()

　　setTimeout(tick.bind(this), 1000)

　　}

　　但組件并不一定只有一份實例，這樣，整個界面上可能就有很多定時器在同時跑，這是一種浪費。如果要做優化，可以把定時器做成一種服務，把業務上需要周期執行的東西放進去，當作定時任務來跑。

　　如果使用RxJS，可以很容易做到這件事：

　　Observable.interval(1000).subscribe(() => {

　　this.diff = moment(createAt).fromNow()

　　})

　　示例二：對時間軸的操縱

　　RxJS一個很強大的特點是，它以流的方式來對待數據，因此，可以用一些操作符對整個流上所有的數據進行延時、取樣、調整密集度等等。

　　const timeA$ = Observable.interval(1000)

　　const timeB$ = timeA$.filter(num => {

　　return (num % 2 != 0)

　　&& (num % 3 != 0)

　　&& (num % 5 != 0)

　　&& (num % 7 != 0)

　　})

　　const timeC$ = timeB$.debounceTime(3000)

　　const timeD$ = timeC$.delay(2000)

　　示例代碼中，我們創建了四個流：

　　A是由定時器產生的，每秒一個值；

　　B從A里面過濾掉了一些；

　　C在B的基礎上，對每兩個間距在3秒之內的值進行了處理，只留下后一個值；

　　D把C的結果整體向后平移了2秒。

　　所以結果大致如下：

　　A: 0  1  2  3  4  5  6  7  8  9  10 11 12 13 14 15 16 17 18 19 20 21

　　B:    1                             11    13          17    19

　　C:          1                                   13                19

　　D:                1                                   13

　　示例三：我們來晚了

　　RxJS還提供了BehaviourSubject和ReplaySubject這樣的東西，用于記錄數據流上一些比較重要的信息，讓那些“我們來晚了”的訂閱者們回放之前錯過的一切。

　　ReplaySubject可以指定保留的值的個數，超過的部分會被丟棄。

　　最近新版《射雕英雄傳》比較火，我們來用代碼描述其中一個場景。

　　郭靖和黃蓉一起背書，黃蓉記憶力很好，看了什么，就全部記得；而郭靖屬魚的，記憶只有七秒，始終只記得背誦的最后三個字，兩人一起背誦《九陰真經》。

　　代碼實現如下：

　　const 九陰真經 = '天之道，損有余而補不足'

　　const 黃蓉$ = new ReplaySubject(Number.MAX_VALUE)

　　const 郭靖$ = new ReplaySubject(3)

　　const 讀書$ = Observable.from(九陰真經.split(''))

　　讀書$.subscribe(黃蓉$)

　　讀書$.subscribe(郭靖$)

　　執行之后，我們就可以看到，黃蓉背出了所有字，郭靖只記得“補不足”三個字。

　　示例四：自動更新的狀態樹

　　熟悉Redux的人應該會對這樣一套理念不陌生：

　　當前視圖狀態 := 之前的狀態 + 本次修改的部分

　　從一個應用啟動之后，整個全局狀態的變化，就等于初始的狀態疊加了之后所有action導致的狀態修改結果。

　　所以這就是一個典型的reduce操作。在RxJS里面，有一個scan操作符可以用來表達這個含義，比如說，我們可以表達這樣一個東西：

　　const action$ = new Subject()

　　const reducer = (state, payload) => {

　　// 把payload疊加到state上返回

　　}

　　const state$ = action$.scan(reducer)

　　.startWith({})

　　只需往這個action$里面推action，就能夠在state$上獲取出當前狀態。

　　在Redux里面，會有一個東西叫combineReducer，在state比較大的時候，用不同的reducer修改state的不同的分支，然后合并。如果使用RxJS，也可以很容易表達出來：

　　const meAction$ = new Subject()

　　const meReducer = (state, payload) => {}

　　const articleAction$ = new Subject()

　　const articleReducer = (state, payload) => {}

　　const me$ = meAction$.scan(meReducer).startWith({})

　　const article$ = articleAction$.scan(articleReducer).startWith({})

　　const state$ = Observable

　　.zip(

　　me$,

　　article$,

　　(me, article) => {me, article}

　　)

　　借助這樣的機制，我們實現了Redux類似的功能，社區里面也有基于RxJS實現的Redux-Observable這樣的Redux中間件。

　　注意，我們這里的代碼中，并未使用dispatch action這樣的方式去嚴格模擬Redux。

　　再深入考慮，在比較復雜的場景下，reducer其實很復雜。比如說，視圖上發起一個操作，會需要修改視圖的好多地方，因此也就是要修改全局狀態樹的不同位置。

　　在這樣的場景中，從視圖發起的某個action，要么調用一個很復雜的reducer去到處改數據，要么再次發起多個action，讓很多個reducer各自改自己的數據。

　　前者的問題是，代碼耦合太嚴重；后者的問題是，整個流程太難追蹤，比如說，某一塊狀態，想要追蹤到自己是被從哪里發起的修改所改變的，是非常困難的事情。

　　如果我們能夠把Observable上面的同步修改過程視為reducer，就可以從另外一些角度大幅簡化代碼，并且讓聯動邏輯清晰化。例如，如果我們想描述一篇文章的編輯權限：

　　const editable$ = Observable.combineLatest(article$, me$)

　　.map(arr => {

　　let [article, me] = arr

　　return me.isAdmin || article.author === me.id

　　})

　　這段代碼的實質是什么？其實本質上還是reducer，表達的是數據的合并與轉換過程，而且是同步的。我們可以把article和me的變更reduce到article$和me$里，由它們派發隱式的action去推動editable計算新值。

　　更詳細探索的可以參見之前的這篇文章：復雜單頁應用的數據層設計。

　　示例五：幸福人生

　　人生是什么樣子的呢？

　　著名央視主持人白巖松曾經說過：

　　賺錢是為了買房，買房是為了賺錢。

　　這兩句話聽上去很悲哀，卻很符合社會現實。（不要在意是不是白巖松說的啦，不是他就是魯迅，要么就是莎士比亞）

　　作為程序員，我們可以嘗試想想如何用代碼把它表達出來。

　　如果用命令式編程的理念來描述這段邏輯，是不太好下手的，因為它看起來像個死循環，可是人生不就是一天一天的死循環嗎，這個復雜的世界，誰是自變量，誰是因變量？

　　死循環之所以很難用代碼表達，是因為你不知道先定義哪個變量，如果變量的依賴關系形成了閉環，就總有一段定義不起來。

　　但是，在RxJS這么一套東西中，我們可以很容易把這套關系描述出來。前面說過，基于RxJS編程，就好像是在組裝管道，依賴關系其實是定義在管道上，而不是在數據上。所以，不存在命令式的那些問題，只要管道能夠接起來，再放進去數據就可以了。所以，我們可以先定義管道之間的依賴關系，

　　首先，從這段話中尋找一些變量，得到如下結果：

　　錢

　　房

　　然后，我們來探索它們各自的來源。

　　錢從哪里來？

　　出租房子。

　　房子從哪里來？

　　錢掙夠了就買。

　　聽上去還是死循環啊？

　　我們接著分析：

　　錢是只有一個來源嗎？

　　不是，原始積累肯定不是房租，我們假定那是工資。

　　所以，收入是有工資和房租兩個部分組成。 房子是只有一個來源嗎？

　　對，我們不是貪官，房子都是用錢買的。

　　好，現在我們有四個變量了：

　　錢

　　房

　　工資

　　房租

　　我們嘗試定義這些變量之間的關系：

　　工資 := 定時取值的常量

　　房租 := 定時取值的變量，與房子數量成正比

　　錢 := 工資 + 房租

　　房 := 錢.map(夠了就買)

　　調整這些變量的定義順序，凡是不依賴別人的，一律提到最前面實現。尷尬地發現，這四個變量里，只有工資是一直不變的，先提前。

　　const salary$ = Observable.interval(100).mapTo(2)

　　剩下的，都是依賴別人的，而且，沒有哪個東西是只依賴已定義的變量，在存在業務上的循環依賴的時候，就會發生這樣的情況。在這種情況下，我們可以從中找出被依賴最少的變量，聲明一個Subject用于占位，比如這里的房子。

　　const house$ = new Subject()

　　接下來再看，以上幾個變量中，有哪個可以跟著確定？是房租，所以，我們可以得到房租與房子數量的關系表達式，注意，以上的salary$、house$，表達的都是單次增加的值，不代表總的值，但是，算房租是要用總的房子數量來算的，所以，我們還需要先表達出總的房子數量：

　　const houseCount$ = house$.scan((acc, num) => acc + num, 0).startWith(0)

　　然后，可以得到房租的表達式：

　　const rent$ = Observable.interval(3000)

　　.withLatestFrom(houseCount$)

　　.map(arr => arr[1] * 5)

　　解釋一下上面這段代碼：

　　房租由房租周期的定時器觸發；

　　然后到房子數量中取最后一個值，也就是當前有多少套房；

　　然后，用房子數量乘以單套房的月租，假設是5；

　　房租定義出來了之后，錢就可以被定義了：

　　const income$ = Observable.merge(salary$, rent$)

　　注意，income$所代表的含義是，所有的單次收入，包含工資和房租。

　　到目前為止，我們還有一個東西沒有被定義，那就是房子。如何從收入轉化為房子呢？為了示例簡單，我們把它們的關系定義為：

　　一旦現金流夠買房，就去買。

　　所以，我們需要定義現金流與房子數量的關系：

　　const cash$ = income$

　　.scan((acc, num) => {

　　const newSum = acc + num

　　const newHouse = Math.floor(newSum / 100)

　　if (newHouse > 0) {

　　house$.next(newHouse)

　　}

　　return newSum % 100

　　}, 0)

　　這段邏輯的含義是：

　　累積之前的現金流與本次收入；

　　假定房價100，先看看現金夠買幾套房，能買幾套買幾套；

　　重新計算買完之后的現金。

　　總結一下，這么一段代碼，就表達清楚了我們所有的業務需求：

　　// 掙錢是為了買房，買房是為了賺錢

　　const house$ = new Subject()

　　const houseCount$ = house$.scan((acc, num) => acc + num, 0).startWith(0)

　　// 工資始終不漲

　　const salary$ = Observable.interval(100).mapTo(2)

　　const rent$ = Observable.interval(3000)

　　.withLatestFrom(houseCount$)

　　.map(arr => arr[1] * 5)

　　// 一買了房，就沒現金了……

　　const income$ = Observable.merge(salary$, rent$)

　　const cash$ = income$

　　.scan((acc, num) => {

　　const newSum = acc + num

　　const newHouse = Math.floor(newSum / 100)

　　if (newHouse > 0) {

　　house$.next(newHouse)

　　}

　　return newSum % 100

　　}, 0)

　　// houseCount$.subscribe(num => console.log(`houseCount: ${num}`))

　　// cash$.subscribe(num => console.log(`cash: ${num}`))

　　這段代碼所表達出來的業務關系如圖：

　　工資周期  ———>  工資

　　↓

　　房租周期  ———>  租金  ———>  收入  ———>  現金

　　↑           ↓

　　房子數量 <——— 新購房

　　注意：在這個例子中，house$的處理方式與眾不同，因為我們的業務邏輯是環形依賴，至少要有一個東西先從里面拿出來占位，后續再處理，否則沒有辦法定義整條鏈路。

　　小結

　　本篇通過一些簡單例子介紹了RxJS的使用場景，可以用這么一句話來描述它：

　　其文簡，其意博，其理奧，其趣深

　　RxJS提供大量的操作符，用于處理不同的業務需求。對于同一個場景來說，可能實現方式會有很多種，需要在寫代碼之前仔細斟酌。由于RxJS的抽象程度很高，所以，可以用很簡短代碼表達很復雜的含義，這對開發人員的要求也會比較高，需要有比較強的歸納能力。

　　本文是入職螞蟻金服之后，第一次內部分享，科普為主，后面可能會逐步作一些深入的探討。

　　螞蟻的大部分業務系統前端不太適合用RxJS，大部分是中后臺CRUD系統，因為兩個原因：整體性、實時性的要求不高。

　　什么是整體性？這是一種系統設計的理念，系統中的很多業務模塊不是孤立的，比如說，從展示上，GUI與命令行的差異在于什么？在于數據的冗余展示。我們可以把同一份業務數據以不同形態展示在不同視圖上，甚至在PC端，由于屏幕大，可以允許同一份數據以不同形態同時展現，這時候，為了整體協調，對此數據的更新就會要產生很多分發和聯動關系。

　　什么是實時性？這個其實有多個含義，一個比較重要的因素是服務端是否會主動向推送一些業務更新信息，如果用得比較多，也會產生不少的分發關系。

　　在分發和聯動關系多的時候，RxJS才能更加體現出它比Generator、Promise的優勢。