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

　　RxJS簡介

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

　　Reactive

　　Lodash for events

　　Observable

　　Stream-based

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

　　比如說，abc三個變量之間存在加法關(guān)系：

　　a = b + c

　　在傳統(tǒng)方式下，這是一種一次性的賦值過程，調(diào)用一次就結(jié)束了，后面b和c再改變，a也不會變了。

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

　　a: = b + c

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

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

　　回調(diào)

　　事件

　　Promise

　　Generator

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

　　分發(fā)

　　流程

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

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

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

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

　　用戶操作

　　網(wǎng)絡(luò)響應

　　定時器

　　Worker

　　RxJS提供了各種API來創(chuàng)建數(shù)據(jù)流：

　　單值：of, empty, never

　　多值：from

　　定時：interval, timer

　　從事件創(chuàng)建：fromEvent

　　從Promise創(chuàng)建：fromPromise

　　自定義創(chuàng)建：create

　　創(chuàng)建出來的數(shù)據(jù)流是一種可觀察的序列，可以被訂閱，也可以被用來做一些轉(zhuǎn)換操作，比如：

　　改變數(shù)據(jù)形態(tài)：map, mapTo, pluck

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

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

　　累加：reduce, scan

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

　　條件執(zhí)行：takeUntil, delayWhen, retryWhen, subscribeOn, ObserveOn

　　轉(zhuǎn)接：switch

　　也可以對若干個數(shù)據(jù)流進行組合：

　　concat，保持原來的序列順序連接兩個數(shù)據(jù)流

　　merge，合并序列

　　race，預設(shè)條件為其中一個數(shù)據(jù)流完成

　　forkJoin，預設(shè)條件為所有數(shù)據(jù)流都完成

　　zip，取各來源數(shù)據(jù)流最后一個值合并為對象

　　combineLatest，取各來源數(shù)據(jù)流最后一個值合并為數(shù)組

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

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

　　那么，數(shù)據(jù)的管道是什么形狀的？

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

　　Observable 可觀察序列，只出不進

　　Observer 觀察者，只進不出

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

　　ReplaySubject 帶回放

　　Subscription 訂閱關(guān)系

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

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

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

　　示例一：簡單的訂閱

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

　　const diff = moment(createAt).fromNow()

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

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

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

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

　　tick() {

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

　　setTimeout(tick.bind(this), 1000)

　　}

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

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

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

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

　　})

　　示例二：對時間軸的操縱

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

　　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)

　　示例代碼中，我們創(chuàng)建了四個流：

　　A是由定時器產(chǎn)生的，每秒一個值；

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

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

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

　　所以結(jié)果大致如下：

　　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這樣的東西，用于記錄數(shù)據(jù)流上一些比較重要的信息，讓那些“我們來晚了”的訂閱者們回放之前錯過的一切。

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

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

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

　　代碼實現(xiàn)如下：

　　const 九陰真經(jīng) = '天之道，損有余而補不足'

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

　　const 郭靖$ = new ReplaySubject(3)

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

　　讀書$.subscribe(黃蓉$)

　　讀書$.subscribe(郭靖$)

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

　　示例四：自動更新的狀態(tài)樹

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

　　當前視圖狀態(tài) := 之前的狀態(tài) + 本次修改的部分

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

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

　　const action$ = new Subject()

　　const reducer = (state, payload) => {

　　// 把payload疊加到state上返回

　　}

　　const state$ = action$.scan(reducer)

　　.startWith({})

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

　　在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}

　　)

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

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

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

　　在這樣的場景中，從視圖發(fā)起的某個action，要么調(diào)用一個很復雜的reducer去到處改數(shù)據(jù)，要么再次發(fā)起多個action，讓很多個reducer各自改自己的數(shù)據(jù)。

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

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

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

　　.map(arr => {

　　let [article, me] = arr

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

　　})

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

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

　　示例五：幸福人生

　　人生是什么樣子的呢？

　　著名央視主持人白巖松曾經(jīng)說過：

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

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

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

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

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

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

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

　　錢

　　房

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

　　錢從哪里來？

　　出租房子。

　　房子從哪里來？

　　錢掙夠了就買。

　　聽上去還是死循環(huán)啊？

　　我們接著分析：

　　錢是只有一個來源嗎？

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

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

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

　　好，現(xiàn)在我們有四個變量了：

　　錢

　　房

　　工資

　　房租

　　我們嘗試定義這些變量之間的關(guān)系：

　　工資 := 定時取值的常量

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

　　錢 := 工資 + 房租

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

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

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

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

　　const house$ = new Subject()

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

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

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

　　const rent$ = Observable.interval(3000)

　　.withLatestFrom(houseCount$)

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

　　解釋一下上面這段代碼：

　　房租由房租周期的定時器觸發(fā)；

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

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

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

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

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

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

　　一旦現(xiàn)金流夠買房，就去買。

　　所以，我們需要定義現(xiàn)金流與房子數(shù)量的關(guān)系：

　　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)

　　這段邏輯的含義是：

　　累積之前的現(xiàn)金流與本次收入；

　　假定房價100，先看看現(xiàn)金夠買幾套房，能買幾套買幾套；

　　重新計算買完之后的現(xiàn)金。

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

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

　　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)

　　// 一買了房，就沒現(xiàn)金了……

　　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}`))

　　這段代碼所表達出來的業(yè)務關(guān)系如圖：

　　工資周期  ———>  工資

　　↓

　　房租周期  ———>  租金  ———>  收入  ———>  現(xiàn)金

　　↑           ↓

　　房子數(shù)量 <——— 新購房

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

　　小結(jié)

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

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

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

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

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

　　什么是整體性？這是一種系統(tǒng)設(shè)計的理念，系統(tǒng)中的很多業(yè)務模塊不是孤立的，比如說，從展示上，GUI與命令行的差異在于什么？在于數(shù)據(jù)的冗余展示。我們可以把同一份業(yè)務數(shù)據(jù)以不同形態(tài)展示在不同視圖上，甚至在PC端，由于屏幕大，可以允許同一份數(shù)據(jù)以不同形態(tài)同時展現(xiàn)，這時候，為了整體協(xié)調(diào)，對此數(shù)據(jù)的更新就會要產(chǎn)生很多分發(fā)和聯(lián)動關(guān)系。

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

　　在分發(fā)和聯(lián)動關(guān)系多的時候，RxJS才能更加體現(xiàn)出它比Generator、Promise的優(yōu)勢。