這是一篇給新手的 RxJS 快速入門，它可能不精確、不全面，但力求對新手友好。

異步與“回調地獄”

我們都知道 JavaScript 是個多范式語言，它既支持過程式編程，又支持函數式編程，兩者分別適用于不同的場合。在同步環境下，兩者各有優缺點，甚至有時候過程式會更簡明一些，但在異步環境下（最典型的場景是一個 Ajax 請求完成后緊接著執行另一個 Ajax 請求），由于無法控制執行和完成的順序，所以就無法使用傳統的過程式寫法，函數式就會展現出其優勢。

問題在于，傳統的函數式寫法實在太不友好了。

傳統寫法下，當我們調用一個 Ajax 時，就要給它一個回調函數，這樣當 Ajax 完成時，就會調用它。當邏輯簡單的時候，這毫無問題。但是我要串起 10 個 Ajax 請求時該怎么辦呢？十重嵌套嗎？恩？似乎有點不對勁兒！

這就是回調地獄。

不僅如此，有時候我到底需要串起多少個 Ajax 請求是未知的，要串起哪些也同樣是未知的。這已經不再是地獄，而是《Mission: Impossible》了。

我，承諾（Promise），幫你解決

事實上，這樣的問題早在 1976 年就已經被發現并解決了。注意，我沒寫錯，確實是 1976 年。

承諾，英文是 Promise [?prɑm?s]，它的基本思想是借助一個代表回執的變量來把回調地獄拍平。

我們以購物為例來看看日常生活中的承諾。

你去電商平臺下單，并付款

平臺會給你一個訂單號，這個訂單號本質上是一個回執，代表商家做出了“稍后我將給你發貨”的承諾

商家發貨給你，在這個過程中你不用等待（異步）

過一段時間，快遞到了

你簽收（回調函數被調用）商品（回調參數）

這次承諾結束

這是最直白的單步驟回調，如果理解了它，再繼續往下看。

你跟電商下的單，但是卻從快遞（并不屬于商家）那里接收到了商品，仔細想想，你不覺得奇怪嗎？雖然表面看確實是商家給你的商品，但我們分解開中間步驟就會發現還有一些幕后的步驟。

商家把商品交給快遞公司，給快遞公司一個訂單號（老的回執）并拿回一個運單號（新的回執）

快遞公司執行這個新承諾，這個過程中商家不用等待（異步）

快遞公司完成這個新承諾，你收到這個新承諾攜帶的商品

所以，事實上，這個購物流程包括兩個承諾：

商家對你的一個發貨承諾

快遞公司對商家的運貨承諾

因此，只要把這些承諾串起來，這些異步動作也就同樣串起來了。

當我們把每個承諾都抽象成一個對象時，我們就可以對任意數量、任意順序的承諾進行組合，變成一個新的承諾。因此回調地獄不復存在，前述的 Mission 也變得 Possible 了。

Promise 的缺點

Promise 固然是一個重大的進步，但在有些場景下仍然是不夠的。比如，Promise 的特點是無論有沒有人關心它的執行結果，它都會立即開始執行，并且你沒有機會取消這次執行。顯然，在某些情況下這么做是浪費的甚至錯誤的。仍然以電商為例，如果某商戶的訂單不允許取消，你還會去買嗎？再舉個編程領域的例子：如果你發起了一個 Ajax 請求，然后用戶導航到了另一個路由，顯然，你這個請求如果還沒有完成就應該被取消，而不應該發出去。但是使用 Promise，你做不到，不是因為實現方面的原因，而是因為它在概念層（接口定義上）就無法支持取消。

此外，由于 Promise 只會承載一個值，因此當我們要處理的是一個集合的時候就比較困難了。比如對于一個隨機數列（總數未知），如果我們要借助 Web API 檢查每個數字的有效性，然后對前一百個有效數字進行求和，那么用 Promise 寫就比較麻煩了。

我們需要一個更高級的 Promise。

Observable

它就是可觀察對象（Observable [?b?z?rv?bl]），Observable 顧名思義就是可以被別人觀察的對象，當它變化時，觀察者就可以得到通知。換句話說，它負責生產數據，別人可以消費它生產的數據。

如果你是個資深后端，那么可能還記得 MessageQueue 的工作模式，它們很像。如果不懂 MQ 也沒關系，我還是用日常知識給你打個比方。

Observable 就像個傳送帶。這個傳送帶不斷運行，圍繞這個傳送帶建立了一條生產線，包括一系列工序，不同的工序承擔單一而確定的職責。每個工位上有一個工人。

整個傳送帶的起點是原料箱，原料箱中的原料不斷被放到傳送帶上。工人只需要待在自己的工位上，對面前的原料進行加工，然后放回傳送帶上或放到另一條傳送帶上即可，簡單、高效、無意外 —— 符合程序員的審美。

而且這個生產線還非常先進 —— 不接單就不生產，非常有效地杜絕了浪費。

FRP

這種設計，看上去很美，對吧？但光看著漂亮可不行，在編程時要怎么實現呢？實際上，這是一種編程范式，叫做函數響應式編程（FRP）。它比 Promise 可年輕多了，直到 1997 年才被人提出來。

顧名思義，FRP 同時具有函數式編程和響應式編程的特點。響應式編程是什么呢？形象的說，它的工作模式就是“飯來張口，衣來伸手”，也就是說，等待外界的輸入，并做出響應。流水線每個工位上的工人正是這種工作模式。

工業上，流水線是人類管理經驗的結晶，它所做的事情是什么呢？本質上就是把每個處理都局部化，以減小復雜度（降低對工人素質的要求）。而這，正是軟件行業所求之不得的。響應式，就是編程領域的流水線。

那么函數式呢？函數式最顯著的特征就是沒有副作用，而這恰好是對流水線上每個工序的要求。顯然，如果某個工序的操作會導致整個生產線平移 10 米，那么用不了多久這個生產線就要掉到海里了，這樣的生產線毫無價值。

因此，響應式和函數式幾乎是注定要在一起的。

ReactiveX

2012 年，微軟 .NET 開發組的一個團隊為了給 LinQ 設計擴展機制而引入了 FRP 概念，卻發現 FRP 的價值不止于此。于是一個新的項目出現了，它就是 ReactiveX。

嚴格來說 ReactiveX 應該是一組 FRP 庫，因為它幾乎在每個主流語言下都提供了實現，而且這些實現都是語言原生風格的，不是簡單地遷移。如果你在任何語言下用過帶有 Rx 前綴的庫，那多半兒就是 ReactiveX 的一個實現了，如 RxJava、Rx.NET、RxGroovy、RxSwift 等等。

ReactiveX 本身其實并不難，難的是 FRP 編程范式以及對操作符（operator）的理解。所以，只要學會了任何一個 Rx* 庫，那么其它語言的庫就可以觸類旁通了。

寶石圖

為了幫助開發者更容易地理解 ReactiveX 的工作原理，ReactiveX 開發組還設計了一種很形象的圖，那就是寶石圖。這貨長這樣（英文注釋不必細看，接下來我會簡單解釋下）：

?

image

?

中間的帶箭頭的線就像傳送帶，用來表示數據序列，這個數據序列被稱為“流”。上方的流叫做輸入流，下方的流叫做輸出流。輸入流可能有多個，但是輸出流只會有一個（不過，流中的每個數據項也可以是別的流）。

數據序列上的每個圓圈表示一個數據項，圓圈的位置表示數據出現的先后順序，但是一般不會表示精確的時間比例，比如在一毫秒內接連出現的兩個數據之間仍然會有較大的距離。只有少數涉及到時間的操作，其寶石圖才會表現出精確的時間比例。

圓圈的最后，通常會有一條豎線或者一個叉號。豎線表示這個流正常終止了，也就是說不會再有更多的數據提供出來了。而叉號表示這個流拋出錯誤導致異常中止了。還有一種流，既沒有豎線也沒有叉號，這種叫做無盡流，比如一個由所有自然數組成的流就不會主動終止。但是要注意，無盡流仍然是可以處理的，因為需要多少項是由消費者決定的。你可以把這個“智能”傳送帶理解為由下一個工位“叫號”的，沒“叫號”下一項數據就不會過來。

中間的大方框表示一個操作，也就是 operator —— 一個函數，比如這個圖中的操作就是把輸入流中的條目乘以十后放入輸出流中。

看懂了寶石圖，就能很形象的理解各種操作符了。

RxJS

主角登場了。RxJS 就是 ReactiveX 在 JavaScript 語言上的實現。對于 JavaScript 程序員來說，不管你是前端還是 NodeJS 后端，RxJS 都會令你受益。

由于 JavaScript 本身的缺陷，RxJS 不得不采用了很多怪異的寫法。它對于 Java / C# 等背景的程序員來說可能會顯得比較怪異，不過，你可以先忽略它們，聚焦在編程范式和接下來要講的操作符語義上。

典型的寫法

of(1,2,3).pipe(filter(item=>item % 2 === 1),map(item=>item * 3), ).subscribe(item=> console.log(item)) `</pre>它會輸出： <pre>`3 9

其中 of 稱為創建器（creator），用來創建流，它返回一個 Observable 類型的對象，filter 和 map 稱為操作符（operator），用來對條目進行處理。這些操作符被當作 Observable 對象的 pipe 方法的參數傳進去。誠然，這個寫法略顯怪異，不過這主要是被 js 的設計缺陷所迫，它已經是目前 js 體系下多種解決方案中相對好看的一種了。

Observable 對象的 subscribe 方法表示消費者要訂閱這個流，當流中出現數據時，傳給 subscribe 方法的回調函數就會被調用，并且把這個數據傳進去。這個回調函數可能被調用很多次，取決于這個流中有多少條數據。

注意，Observable 必須被 subscribe 之后才會開始生產數據。如果沒人 subscribe 它，那就什么都不會做。

簡單創建器

廣義上，創建器也是操作符的一種，不過這里我們把它單獨拿出來講。要啟動生產線，我們得先提供原料。本質上，這個提供者就是一組函數，當流水線需要拿新的原料時，就會調用它。

你當然可以自己實現這個提供者，但通常是不用的。RxJS 提供了很多預定義的創建器，而且將來可能還會增加新的。不過，那些眼花繚亂的創建器完全沒必要全記住，只要記住少數幾個就夠了，其它的有時間慢慢看。

of - 單一值轉為流

?

image

?

它接收任意多個參數，參數可以是任意類型，然后它會把這些參數逐個放入流中。

from - 數組轉為流

?

image

?

它接受一個數組型參數，數組中可以有任意數據，然后把數組的每個元素逐個放入流中。

range - 范圍轉為流

?

image

?

它接受兩個數字型參數，一個起點，一個終點，然后按 1 遞增，把中間的每個數字（含邊界值）放入流中。

fromPromise - Promise 轉為流

接受一個 Promise，當這個 Promise 有了輸出時，就把這個輸出放入流中。

要注意的是，當 Promise 作為參數傳給 fromPromise 時，這個 Promise 就開始執行了，你沒有機會防止它被執行。

如果你需要這個 Promise 被消費時才執行，那就要改用接下來要講的 defer 創建器。

defer - 惰性創建流

?

image

?

它的參數是一個用來生產流的工廠函數。也就是說，當消費方需要流（注意不是需要流中的值）的時候，就會調用這個函數，創建一個流，并從這個流中進行消費（取數據）。

因此，當我們定義 defer 的時候，實際上還不存在一個真正的流，只是給出了創建這個流的方法，所以叫惰性創建流。

timer - 定時器流

?

image

?

它有兩個數字型的參數，第一個是首次等待時間，第二個是重復間隔時間。從圖上可以看出，它實際上是個無盡流 —— 沒有終止線。因此它會按照預定的規則往流中不斷重復發出數據。

要注意，雖然名字有相關性，但它不是 setTimeout 的等價物，事實上它的行為更像是 setInterval。

interval - 定時器流

?

image

?

它和 timer 唯一的差別是它只接受一個參數。事實上，它就是一個語法糖，相當于 timer(1000, 1000)，也就是說初始等待時間和間隔時間是一樣的。

如果需求確實是 interval 的語義，那么就優先使用這個語法糖，畢竟，從行為上它和 setInterval 幾乎是一樣的。

思考題：假設點了一個按鈕之后我要立刻開始一個動作，然后每隔 1000 毫秒重復一次，該怎么做？換句話說：該怎么移除首次延遲時間？

Subject - 主體對象

它和創建器不同，創建器是供直接調用的函數，而 Subject 則是一個實現了 Observable 接口的類。也就是說，你要先把它 new 出來（假設實例叫 subject），然后你就可以通過程序控制的方式往流里手動放數據了。它的典型用法是用來管理事件，比如當用戶點擊了某個按鈕時，你希望發出一個事件，那么就可以調用 subject.next(someValue) 來把事件內容放進流中。

當你希望手動控制往這個流中放數據的時機時，這種特性非常有用。

當然，Subject 其實并沒有這么簡單，用法也很多，不過這部分內容超出了本文的范圍。

合并創建器

我們不但可以直接創建流，還可以對多個現有的流進行不同形式的合并，創建一個新的流。常見的合并方式有三種：并聯、串聯、拉鏈。

merge - 并聯

?

image

?

從圖上我們可以看到兩個流中的內容被合并到了一個流中。只要任何一個流中出現了值就會立刻被輸出，哪怕其中一個流是完全空的也不影響結果 —— 等同于原始流。

這種工作方式非常像電路中的并聯行為，因此我稱其為并聯創建器。

并聯在什么情況下起作用呢？舉個例子吧：有一個列表需要每隔 5 秒鐘定時刷新一次，但是一旦用戶按了搜索按鈕，就必須立即刷新，而不能等待 5 秒間隔。這時候就可以用一個定時器流和一個自定義的用戶操作流（subject）merge 在一起。這樣，無論哪個流中出現了數據，都會進行刷新。

concat - 串聯

?

image

?

從圖中我們可以看到兩個流中的內容被按照順序放進了輸出流中。前面的流尚未結束時（注意豎線），后面的流就會一直等待。

這種工作方式非常像電路中的串聯行為，因此我稱其為串聯創建器。

串聯的適用場景就很容易想象了，比如我們需要先通過 Web API 進行登錄，然后取學生名冊。這兩個操作就是異步且串聯工作的。

zip - 拉鏈

?

image

?

zip 的直譯就是拉鏈，事實上，有些壓縮軟件的圖標就是一個帶拉鏈的鑰匙包。拉鏈的特點是兩邊各有一個“齒”，兩者會嚙合在一起。這里的 zip 操作也是如此。

從圖上我們可以看到，兩個輸入流中分別出現了一些數據，當僅僅輸入流 A 中出現了數據時，輸出流中什么都沒有，因為它還在等另一個“齒”。當輸出流 B 中出現了數據時，兩個“齒”都湊齊了，于是對這兩個齒執行中間定義的運算（取 A 的形狀，B 的顏色，并合成為輸出數據）。

可以看到，當任何一個流先行結束之后，整個輸出流也就結束了。

拉鏈創建器適用的場景要少一些，通常用于合并兩個數據有對應關系的數據源。比如一個流中是姓名，另一個流中是成績，還有一個流中是年齡，如果這三個流中的每個條目都有精確的對應關系，那么就可以通過 zip 把它們合并成一個由表示學生成績的對象組成的流。

操作符

RxJS 有很多操作符，事實上比創建器還要多一些，但是我們并不需要一一講解，因為它們中的很大一部分都是函數式編程中的標配，比如 map、reduce、filter 等。有 Java 8 / scala / kotlin 等基礎的后端或者用過 underscore/lodash 的前端都可以非常容易地理解它們。

本文重點講解一些傳統方式下沒有的或不常用的：

retry - 失敗時重試

?

image

?

有些錯誤是可以通過重試進行恢復的，比如臨時性的網絡丟包。甚至一些流程的設計還會故意借助重試機制，比如當你發起請求時，如果后端發現你沒有登錄過，就會給你一個 401 錯誤，然后你可以完成登錄并重新開始整個流程。

retry 操作符就是負責在失敗時自動發起重試的，它可以接受一個參數，用來指定最大重試次數。

這里我為什么一直在強調失敗時重試呢？因為還有一個操作符負責成功時重試。

repeat - 成功時重試

?

image

?

除了重復的條件之外，repeat 的行為幾乎和 retry 一模一樣。

repeat 很少會單獨用，一般會組合上 delay 操作，以提供暫停時間，否則就容易 DoS 了服務器。

delay - 延遲

?

image

?

這才是真正的 setTimeout 的等價操作。它接受一個毫秒數（圖中是 20 毫秒），每當它從輸入流中讀取一個數據之后，會先等待 20 毫秒，然后再放到輸出流中。

可以看到，輸入流和輸出流內容是完全一樣的，只是時機上，輸出流中的每個條目都恰好比輸入流晚 20 毫秒出現。

toArray - 收集為數組

?

image

?

事實上，你幾乎可以把它看做是 from 的逆運算。 from 把數組打散了逐個放進流中，而 toArray 恰好相反，把流中的內容收集到一個數組中 —— 直到這個流結束。

這個操作符幾乎總是放在最后一步，因為 RxJS 的各種 operator 本身就可以對流中的數據進行很多類似數組的操作，比如查找最小值、最大值、過濾等。所以通常會先使用各種 operator 對數據流進行處理，等到要脫離 RxJS 的體系時，再轉換成數組傳出去。

debounceTime - 防抖

?

image

?

在 underscore/lodash 中這是常用函數。 所謂防抖其實就是“等它平靜下來”。比如預輸入（type ahead）功能，當用戶正在快速打字的時候，你沒必要立刻去查服務器，否則可能直接讓服務器掛了，而應該等用戶稍作停頓（平靜下來）時再發起查詢。

debounceTime 就是這樣，你傳入一個最小平靜時間，在這個時間窗口內連續過來的數據一概被忽略，一旦平靜時間超過它，就會往把接收到的下一條數據放到流中。這樣消費者就只能看到平靜時間超時之后發來的最后一條數據。

switchMap - 切換成另一個流

?

image

?

這可能是相對較難理解的一個 operator。

有時候，我們會希望根據一個立即數發起一個遠程查詢，并且把這個異步取回的結果放進流中。比如，流中是一些學生的 id，每過來一個 id，你要發起一個 Ajax 請求來根據這個 id 獲取這個學生的詳情，并且把詳情放進輸出流中。

注意，這是一個異步操作，所以你沒法用普通的 map 來實現，否則映射出來的結果就會是一個個 Observable 對象。

switchMap 就是用來解決這個問題的。它在回調函數中接受從輸入流中傳來的數據，并轉換成一個新的 Observable 對象（新的流，每個流中包括三個值，每個值都等于輸入值的十倍），switchMap 會訂閱這個 Observable 對象，并把它的值放入輸出流中。注意圖中豎線的位置 —— 只有當所有新的流都結束時，輸出流才會結束。

不知道你有沒有注意到這里一個很重要的細節。30 只生成了兩個值，而不是我們所預期的三個。這是因為當輸入流中的 5 到來時，會切換到以 5 為參數構建出的這個新流（S5），而這時候基于 3 構建的那個流（S3）尚未結束。雖然如此，但是已經沒人再訂閱 S3 了，因為同一時刻 switchMap 只能訂閱一個流。所以，已經沒人會再朝著 S3 “叫號”了，它已經被釋放了。

規律：operator 打包學

當你掌握了一些基本操作符之后，就可以讓自己的操作符知識翻倍了。

這是因為 RxJS 中的很多操作符都遵循著同樣的命名模式。比如：

xxxWhen - 滿足條件時 xxx

它接受一個 Observable 型參數作為條件流，一旦這個條件流中出現任意數據，則進行 xxx 操作。

如 retryWhen(notifier$)，其中的 notifier$ 就是一個條件流。當輸入流出現異常時，就會開始等待 notifier$ 流中出現數據，一旦出現了任何數據（不管是什么值），就會開始執行重試邏輯。

xxxCount - 拿到 n 個數據項時 xxx

它接受一個數字型參數作為閾值，一旦從輸入流中取到了 n 個數據，則進行 xxx 操作。

如 bufferCount(3) 表示每拿到 3 個數據就進行一次 buffer 操作。

這個操作可以看做是 xxxWhen 的語法糖。

xxxTime - 超時后 xxx

它接受一個超時時間作為參數，從輸入流中取數據，一旦到達超時時間，則執行 xxx 操作。

比如前面講過的 debounceTime 其實遵循的就是這種模式。

這個操作可以看做 xxxWhen 的語法糖。

xxxTo - 用立即量代替 Lambda 表達式

它接受一個立即量作為參數，相當于 xxx(()=>value))。

比如 mapTo('a') 其實是 map(()=>'a') 的語法糖，也就是說無論輸入流中給出的值是什么，我往輸出流中放入的都是這個固定的值。

坑與最佳實踐

取消訂閱

subscribe 之后，你的回調函數就被別人引用了，因此如果不撤銷對這個回調函數的引用，那么與它相關的內存就永遠不會釋放，同時，它仍然會在流中有數據過來時被調用，可能會導致奇怪的 console.log 等意外行為。

因此，必須找到某個時機撤銷對這個回調函數的引用。但其實不一定需要那么麻煩。解除對回調函數的引用有兩種時機，一種是這個流完成（complete，包括正常結束和異常結束）了，一種是訂閱方主動取消。當流完成時，會自動解除全部訂閱回調，而所有的有限流都是會自動完成的。只有無盡流才需要特別處理，也就是訂閱方要主動取消訂閱。

當調用 Observable 的 subscribe 方法時，會返回一個 Subscription 類型的引用，它實際上是一個訂閱憑證。把它保存下來，等恰當的時機調用它的 unsubscribe 方法就可以取消訂閱了。比如在 Angular 中，如果你訂閱了無盡流，那么就需要把訂閱憑證保存在私有變量里，并且在 ngOnDestroy 回調中調用它的 unsubscribe 方法。

類型檢查

只要有可能，請盡量使用 TypeScript 來書寫 RxJS 程序。由于大量 operator 都會改變流中的數據類型，因此如果靠人力來追蹤數據類型的變化既繁瑣又容易出錯。TypeScript 的類型檢查可以給你提供很大的幫助，既省心又安全，而且這兩個都是微軟家的，搭配使用，風味更佳。

代碼風格

如同所有 FP 程序一樣，ReactiveX 的代碼也應該由一系列小的、單一職責的、無副作用的函數組成。雖然 JavaScript 無法像 Java 中那樣對 Lambda 表達式的副作用做出編譯期限制，但是仍然要遵循同樣的原則，堅持無副作用和數據不變性。

寄語 - 實踐出真知

ReactiveX 大家族看似龐大，實則簡單 —— 如果你已經有了 Java 8+ / Kotlin / underscore 或 lodash 等函數式基礎知識時，新東西就很少了。而當你用過 Rx 大家族中的任何一個成員時，RxJS 對你幾乎是免費的，反之也一樣。

唯一的問題，就是找機會實踐，并體會 FRP 風格的獨特之處，獲得那些超乎具體技術之上的真知灼見。

作者：ThoughtWorks
鏈接：https://www.jianshu.com/p/bc4d8ce267d1
來源：簡書
著作權歸作者所有。商業轉載請聯系作者獲得授權，非商業轉載請注明出處。

總結

以上是生活随笔為你收集整理的RxJS 新手快速入门的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。