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的優勢。

RxJS的基礎