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

　　RxJS簡(jiǎn)介

　　通常，對RxJS的解釋會(huì )是這么一些東西，我們來(lái)分別看看它們的含義是什么。

　　Reactive

　　Lodash for events

　　Observable

　　Stream-based

　　什么是Reactive呢，一個(gè)比較直觀(guān)的對比是這樣的：

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

　　a = b + c

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

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

　　a: = b + c

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

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

　　回調

　　事件

　　Promise

　　Generator

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

　　分發(fā)

　　流程

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

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

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

　　我們可以把一切輸入都當做數據流來(lái)處理，比如說(shuō)：

　　用戶(hù)操作

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

　　定時(shí)器

　　Worker

　　RxJS提供了各種API來(lái)創(chuàng )建數據流：

　　單值：of, empty, never

　　多值：from

　　定時(shí)：interval, timer

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

　　從Promise創(chuàng )建：fromPromise

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

　　創(chuàng )建出來(lái)的數據流是一種可觀(guān)察的序列，可以被訂閱，也可以被用來(lái)做一些轉換操作，比如：

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

　　過(guò)濾一些值：filter, skip, first, last, take

　　時(shí)間軸上的操作：delay, timeout, throttle, debounce, audit, bufferTime

　　累加：reduce, scan

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

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

　　轉接：switch

　　也可以對若干個(gè)數據流進(jìn)行組合：

　　concat，保持原來(lái)的序列順序連接兩個(gè)數據流

　　merge，合并序列

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

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

　　zip，取各來(lái)源數據流最后一個(gè)值合并為對象

　　combineLatest，取各來(lái)源數據流最后一個(gè)值合并為數組

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

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

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

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

　　Observable 可觀(guān)察序列，只出不進(jìn)

　　Observer 觀(guān)察者，只進(jìn)不出

　　Subject 可出可進(jìn)的可觀(guān)察序列，可作為觀(guān)察者

　　ReplaySubject 帶回放

　　Subscription 訂閱關(guān)系

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

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

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

　　示例一：簡(jiǎn)單的訂閱

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

　　const diff = moment(createAt).fromNow()

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

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

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

　　在沒(méi)有RxJS的情況下，我們可能會(huì )通過(guò)一個(gè)定時(shí)器來(lái)做這件事，比如在組件內部：

　　tick() {

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

　　setTimeout(tick.bind(this), 1000)

　　}

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

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

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

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

　　})

　　示例二：對時(shí)間軸的操縱

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

　　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 )建了四個(gè)流：

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

　　B從A里面過(guò)濾掉了一些；

　　C在B的基礎上，對每?jì)蓚€(gè)間距在3秒之內的值進(jìn)行了處理，只留下后一個(gè)值；

　　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

　　示例三：我們來(lái)晚了

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

　　ReplaySubject可以指定保留的值的個(gè)數，超過(guò)的部分會(huì )被丟棄。

　　最近新版《射雕英雄傳》比較火，我們來(lái)用代碼描述其中一個(gè)場(chǎng)景。

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

　　代碼實(shí)現如下：

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

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

　　const 郭靖$ = new ReplaySubject(3)

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

　　讀書(shū)$.subscribe(黃蓉$)

　　讀書(shū)$.subscribe(郭靖$)

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

　　示例四：自動(dòng)更新的狀態(tài)樹(shù)

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

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

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

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

　　const action$ = new Subject()

　　const reducer = (state, payload) => {

　　// 把payload疊加到state上返回

　　}

　　const state$ = action$.scan(reducer)

　　.startWith({})

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

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

　　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}

　　)

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

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

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

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

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

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

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

　　.map(arr => {

　　let [article, me] = arr

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

　　})

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

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

　　示例五：幸福人生

　　人生是什么樣子的呢？

　　著(zhù)名央視主持人白巖松曾經(jīng)說(shuō)過(guò)：

　　賺錢(qián)是為了買(mǎi)房，買(mǎi)房是為了賺錢(qián)。

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

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

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

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

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

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

　　錢(qián)

　　房

　　然后，我們來(lái)探索它們各自的來(lái)源。

　　錢(qián)從哪里來(lái)？

　　出租房子。

　　房子從哪里來(lái)？

　　錢(qián)掙夠了就買(mǎi)。

　　聽(tīng)上去還是死循環(huán)??？

　　我們接著(zhù)分析：

　　錢(qián)是只有一個(gè)來(lái)源嗎？

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

　　所以，收入是有工資和房租兩個(gè)部分組成。 房子是只有一個(gè)來(lái)源嗎？

　　對，我們不是貪官，房子都是用錢(qián)買(mǎi)的。

　　好，現在我們有四個(gè)變量了：

　　錢(qián)

　　房

　　工資

　　房租

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

　　工資 := 定時(shí)取值的常量

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

　　錢(qián) := 工資 + 房租

　　房 := 錢(qián).map(夠了就買(mǎi))

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

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

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

　　const house$ = new Subject()

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

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

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

　　const rent$ = Observable.interval(3000)

　　.withLatestFrom(houseCount$)

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

　　解釋一下上面這段代碼：

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

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

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

　　房租定義出來(lái)了之后，錢(qián)就可以被定義了：

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

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

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

　　一旦現金流夠買(mǎi)房，就去買(mǎi)。

　　所以，我們需要定義現金流與房子數量的關(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)

　　這段邏輯的含義是：

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

　　假定房?jì)r(jià)100，先看看現金夠買(mǎi)幾套房，能買(mǎi)幾套買(mǎi)幾套；

　　重新計算買(mǎi)完之后的現金。

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

　　// 掙錢(qián)是為了買(mǎi)房，買(mǎi)房是為了賺錢(qián)

　　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)

　　// 一買(mǎi)了房，就沒(méi)現金了……

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

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

　　工資周期  ———>  工資

　　↓

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

　　↑           ↓

　　房子數量 <——— 新購房

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

　　小結

　　本篇通過(guò)一些簡(jiǎn)單例子介紹了RxJS的使用場(chǎng)景，可以用這么一句話(huà)來(lái)描述它：

　　其文簡(jiǎn)，其意博，其理奧，其趣深

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

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

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

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

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

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