作者：霍世杰 @阿杰

前言

打開此文的小伙伴想必對 RxJS 已經(jīng)有了或多或少的了解，如果沒有倒也無妨，因為下面會從零開始講起；如果你帶著幾個問題來翻閱，本人也希望此文可以帶你找到答案。

溫馨提示：文章內(nèi)容較長，建議收藏反復(fù)觀看。

概覽

從我個人的學(xué)習(xí) RxJS 的歷程來看，最開始是“照貓畫虎”能夠基本使用，隨后是研究部分操作符和使用場景，最后了解產(chǎn)生背景、設(shè)計思想以及實現(xiàn)原理。在這期間有過很多疑問，也曾從不同角度理解 RxJS，最終總結(jié)了認(rèn)為比較系統(tǒng)的知識圖譜（下圖）。

深入理解 RxJS

大“道”——響應(yīng)式編程

全面理解一個事物，追溯其歷史是一種好的方式，RxJS 的起源需要追溯到響應(yīng)式編程（RP），后續(xù)產(chǎn)生了一系列基于響應(yīng)式編程范式的語言擴展（Rx，RxJS 就是其中之一），請看歷史簡譜（左向右延續(xù)）。

何為響應(yīng)式

響應(yīng)式是學(xué)習(xí) RxJS 必須要理解的概念，本人用了大量的文字來解釋，如果您已經(jīng)深刻理解，可直接跳過。如果您是第一次接觸這個名詞，也不要先給自己心里暗示它有多么的高深和難以理解，也許你天天在使用。

一個例子

為了避免上來就接觸晦澀的概念，先來舉個例子：博客平臺關(guān)注功能。話說你偶然瀏覽到阿杰的文章，覺得寫的很贊，于是你關(guān)注了他的博客賬號，以便不會錯過之后的干貨，在以后的日子里阿杰每發(fā)布一篇文章博客平臺都會給你推送一條消息，提醒你來給他點點贊，假設(shè)博客平臺沒有關(guān)注的功能，那么你需要想知道他的最新動態(tài)就只能打開他的個人主頁查看文章列表來確認(rèn)，也許稍不留意就會錯過他的文章。這個例子出現(xiàn)了粉絲關(guān)注博主、博主發(fā)布博客、平臺自動推送給粉絲消息、給文章點贊，這就形成了響應(yīng)式閉環(huán)，平臺在觀察到博主粉絲只需要關(guān)注一下就能收到博主以后的動態(tài)，這就是響應(yīng)式帶來的好處。

另一個例子

再舉一個貼近我們開發(fā)的例子：假設(shè)有一個更新某用戶密碼的需求，A 同事負(fù)責(zé)調(diào)用更新邏輯并在更新后執(zhí)行其他任務(wù)（比如提醒用戶更新成功或失敗），B 同事負(fù)責(zé)具體更新密碼的邏輯，下圖描述了完成整個任務(wù)的流程：

1. 驗證一下用戶信息的真實性

2. 驗證密碼是否合法

3. 最終把新的密碼入庫

上述的每個環(huán)節(jié)都有可能是異步耗時任務(wù)，比如用戶的真實性是第三方平臺驗證的，入庫的過程中網(wǎng)絡(luò)非常慢，再比如......等等，諸如此類的各種不確定性，這對于 B 同事做后續(xù)任務(wù)就有了一個關(guān)鍵性條件，確定/等待更新結(jié)果，這種情況有一種做法是：定期輪詢重試，B 每隔一段時間執(zhí)行一次，直到確定 A 已經(jīng)修改成功，再去執(zhí)行后續(xù)操作。邏輯中定時 A 邏輯結(jié)束這種做法這種做法明顯有一個弊端是執(zhí)行多次，對于 B 顯然不是好的做法，好的做法是：B 的更新邏輯執(zhí)行完后通知 A，甚至 B 可以先把更新后的事準(zhǔn)備好，讓 A 決定后續(xù)邏輯的執(zhí)行時機。

流程如圖示：訂閱/執(zhí)行更新邏輯、更新邏輯結(jié)束、將結(jié)果通知調(diào)用者、執(zhí)行后續(xù)邏輯。這就是響應(yīng)式的做法，它帶來的好處是：當(dāng)更新結(jié)果發(fā)生變化時自動通知調(diào)用者，而不用輪詢重試。

了解響應(yīng)式宣言

相信你已經(jīng)明白了響應(yīng)式，并能發(fā)現(xiàn)生活/工作中到處可見，下面了解一下設(shè)計響應(yīng)式模塊/系統(tǒng)遵循的原則：

• 即時響應(yīng)性：只要有可能，就要及時地做出響應(yīng)。

• 回彈性：執(zhí)行過程中在出現(xiàn)失敗時依然保持即時響應(yīng)性。

• 彈性： 在不斷變化的工作負(fù)載之下依然保持即時響應(yīng)性。

• 消息驅(qū)動：反應(yīng)式依賴異步的消息傳遞，從而確保了松耦合、隔離、位置透明的組件之間有著明確邊界。

響應(yīng)式編程

下面我們正式的介紹響應(yīng)式編程：

響應(yīng)式編程，Reactive Programing，又稱反應(yīng)式編程，簡稱 RP，是一種以傳播數(shù)據(jù)流（數(shù)據(jù)流概念戳這里 ）的編程范式。

響應(yīng)式編程或反應(yīng)式編程（英語：Reactive programming）是一種面向數(shù)據(jù)流和變化傳播的聲明式編程范式。這意味著可以在編程語言中很方便地表達(dá)靜態(tài)或動態(tài)的數(shù)據(jù)流，而相關(guān)的計算模型會自動將變化的值通過數(shù)據(jù)流進(jìn)行傳播。

優(yōu)勢：

1. 聲明式，方便地表達(dá)靜態(tài)或動態(tài)的數(shù)據(jù)流

2. 自動化，自動將變化的值通過數(shù)據(jù)流進(jìn)行傳播

核心思想：從傳統(tǒng)的調(diào)用方“拉”數(shù)據(jù)的思維模式轉(zhuǎn)變?yōu)楸徽{(diào)用方“推”數(shù)據(jù)的思維模式。

JS 異步編程史

眾所周知，JS 執(zhí)行環(huán)境是單線程的，在事件監(jiān)聽，異步的處理，響應(yīng)式編程毋庸置疑是其中的一大主力。

Callback 時代

回調(diào)函數(shù)延續(xù)至今，JS 的運用高階函數(shù)巧妙地將異步后的邏輯進(jìn)行托管，以事件驅(qū)動的方式來解決異步編程，但它有一個“臭名昭著”的問題：回調(diào)嵌套，耦合度高。本來很簡單的邏輯但為了控制執(zhí)行流程卻不得不寫大量的代碼，當(dāng)時產(chǎn)生了一些知名的庫：async、bluebrid，它們封裝和處理了嵌套問題，暴露出更為簡單好用的 API，額外還可以優(yōu)雅地處理流程控制相關(guān)場景，但所做的只是劃分了邏輯，依舊沒有解決代碼臃腫的問題。

Promise 時代

ES6 納入 Promise 之后可謂一大喜訊，因為它解決了回調(diào)嵌套的問題，雖然它只是回調(diào)的語法糖，但在處理流程和捕獲錯誤（外層處理）已經(jīng)非常的優(yōu)雅了，但它的弊端是：無法監(jiān)聽和打斷 Promise 的狀態(tài)。這意味著一旦聲明它會立即執(zhí)行并修改它的執(zhí)行狀態(tài)，這源于它的實現(xiàn)。

Generator

Generator 是處于 Promise 和 Async/await 之間的產(chǎn)物，它給我們帶來了寫異步語法像寫同步一般，只需在函數(shù)前加 * 修飾，這樣就可以在函數(shù)內(nèi)部使用一個 yield 關(guān)鍵字返回結(jié)果，類似于 await ，但它也并非完美，不然也不會有后面的 Async/await 了，它的主要問題是流程管理不方便（迭代器模式實現(xiàn)，主動調(diào) next 執(zhí)行器流轉(zhuǎn)游標(biāo)）。

Async/await

Async/await 是 Generator 語法糖，既保留了語法上的優(yōu)勢，也解決了 Generator 每步要調(diào)一下 next 執(zhí)行器的弊端，是現(xiàn)階段的最佳方案，就是得吐槽一下 Top-level await 到 ES2022 才出現(xiàn)。

其中 Generator 和 Async/await 在異步編程是以等待的方式處理。

ReactiveX

業(yè)界一致認(rèn)為正統(tǒng)的響應(yīng)式實現(xiàn)/擴展是 ReactiveX 系列。

ReactiveX，簡稱 Rx，是基于響應(yīng)式的擴展，是各種語言實現(xiàn)的一個統(tǒng)稱，除了我們所知道的 RxJS，還有 RxJava、http://Rx.NET、RxKotlin、RxPHP.....它最早是由微軟提出并引入到 .NET 平臺中，隨后 ES6 也引入了類似的技術(shù)。

它擴展了觀察者模式，以支持?jǐn)?shù)據(jù)序列和/或事件，并添加了操作符，允許您以聲明的方式將序列組合在一起，同時抽象出諸如低級線程、同步、線程安全、并發(fā)數(shù)據(jù)結(jié)構(gòu)和非阻塞I/O等問題。

RxJS

RxJS 全稱 Reactive Extensions for JavaScript，翻譯過來是 Javascript 的響應(yīng)式擴展，它是一個采用流來處理異步和事件的工具庫，簡單來說 Rx(JS) = Observables + Operator + Scheduler。

擅長做的事：

• UI 事件：例如鼠標(biāo)移動、按鈕單擊......

• 狀態(tài)管理：例如屬性更改、集合更新等事件

• IO 消息事件：服務(wù)監(jiān)聽

• 廣播/通知：消息總線（Event bus）

• 網(wǎng)絡(luò)消息/事件：例如 HTTP、WebSockets API 或其他低延遲中間件

最大的優(yōu)勢：異步事件的抽象，這意味著可以把很多事統(tǒng)一起來當(dāng)作一種方式處理，從而讓問題變得更簡單，同時也降低了學(xué)習(xí)成本。

注意：RxJS 擅長做異步的事，不代表不可以做同步或不擅長同步的事。

RxJS 在 Angular 中的應(yīng)用

RxJS 在 Angular 中及其重要，很多核心模塊都是由 RxJS 實現(xiàn)的，比如：

• 響應(yīng)式表單

• 路由

• HttpClient（封裝的 ajax，類似于 axios）

• async 管道符

• 狀態(tài)管理

更多： https://angular.io/guide/observables-in-angular

RxJS 核心概念—— Observables

RxJS 中的 Observables 系列是圍繞觀察者模式來實現(xiàn)的，基本角色：

1. Observable：被觀察者，用來產(chǎn)生消息/數(shù)據(jù)。

2. Observer：觀察者，用來消費消息/數(shù)據(jù)。

Observable

Observeable 是觀察者模式中的被觀察者，它維護(hù)一段執(zhí)行函數(shù)，提供了惰性執(zhí)行的能力（subscribe）。

核心函數(shù)

• constructor(_subscribe) : 創(chuàng)建 Observeable

• static create(_subscribe)：靜態(tài)函數(shù)創(chuàng)建 Observeable

• pipe()：管道

• subscribe()：執(zhí)行初始化傳入的 _subscribe

RxJS 中 Observeable 是一等公民，將一切問題都轉(zhuǎn)化為 Observable 去處理。轉(zhuǎn)換的操作符有from 、 fromEvent 、 of 、 timer 等等，更多戳 這里。
注意的是：只有 ObservableInput或 SubscribableOrPromise類型的值才可以轉(zhuǎn)化為 Observable。

基本使用

源碼實現(xiàn)

本人寫（抽?。┝艘惶?RxJS Observable 源碼中的核心實現(xiàn)

Observable 與 Promise

用過兩者的同學(xué)可能會有疑問為什么采用 Observable 而不直接用 Promise 或 Async/await，這兩者在業(yè)界也常常用來做對比。

它們關(guān)鍵性的不同點：

總的來說，Promise 可讀性更優(yōu)，Observable 從使用場景更為全面。

兩者的相互轉(zhuǎn)換

在既使用了 RxJS 又引用了用 Promise 封裝的庫時，兩者相互轉(zhuǎn)換是容易碰到的問題，RxJS 提供了兩者轉(zhuǎn)換的函數(shù)。

Promise 轉(zhuǎn) Observable

from 或 fromPromise（棄用） 操作符

const observable$ = from(fetch('http://xxx.com/'));

Observable 轉(zhuǎn) Promise

const promise = of(42).toPromise();
const errorPromise = throw(new Error('woops')).toPromise();
errorPromise.catch(err=> console.error);

Subscriber/Observer

Subscriber/Observer 是觀察者模式中的觀察者/消費者，它用來消費/執(zhí)行 Observable 創(chuàng)建的函數(shù)。

核心能力

• next （傳值）

• error （錯誤處理）

• complete （完成/終止）

實現(xiàn)

白話描述：

1. 將 subscribe 傳進(jìn)去一個 next 函數(shù)賦給 Observer 的 next 函數(shù)。

2. 將 Observer 傳給 Observable 初始化的預(yù)加載函數(shù) _subscribe。

3. 執(zhí)行 Observable 初始化的預(yù)加載函數(shù)

工作流程

Subscription

上面的 Observable 和 Observer 已經(jīng)完成了觀察者模式的核心能力，但是引發(fā)的一個問題是，每次執(zhí)行一個流創(chuàng)建一個 Observable，這可能會創(chuàng)建多個對象（尤其是大量使用操作符時，會創(chuàng)建多個 Observable 對象，這個我們后面再說），此時需要外部去銷毀此對象，不然會造成內(nèi)存泄露。

為了解決這個問題，所以產(chǎn)生了一個 Subscription 的對象，Subscription 是表示可清理資源的對象，它是由 Observable 執(zhí)行之后產(chǎn)生的。

核心能力

• unsubcribe （取消訂閱）

• add （分組或在取消訂閱之前插入一段邏輯）

注意：調(diào)用unsubcribe后（包含add傳入的其它 Subscription）不會再接收到它們的數(shù)據(jù)。

使用

實現(xiàn)

白話描述：

1. 調(diào)用 Observable 的 subscribe 后會添加（add 方法）到 Subscription（這里有個關(guān)系 Subscriber 繼承了 Subscription） 中，并把 Subscriber（也是 Subscription）返出去。

2. 調(diào)用 Subscription 的 unsubscribe 方法。

3. unsubscribe 把該對象置空回收。

完整工作流程

Subject

上述的 Observable 歸根到底就是一個惰性執(zhí)行的過程，當(dāng)遇到以下兩種情況就顯得偏弱：

1. 推送多條數(shù)據(jù)時，需要就要創(chuàng)建多個對象。

2. 做狀態(tài)管理或消息通訊，監(jiān)聽數(shù)據(jù)變化并實時推送。

基于這兩個方面，所以產(chǎn)生了 Subject，Subject 是一個特殊的 Observable，更像一個 EventEmitter，它既可以是被觀察者/生產(chǎn)者也可以是觀察者/消費者。

優(yōu)勢

1. 減少開銷和提高性能

2. 數(shù)據(jù)實時推送

場景

消息傳遞或廣播。

與 Observable 的區(qū)別

重點解釋一下消費策略和消費時機兩塊：

冷數(shù)據(jù)流：可以訂閱任意時間的數(shù)據(jù)流。

熱數(shù)據(jù)流：只給已訂閱的消費者發(fā)送消息，定閱之前的消費者，不會收到消息。

用一個示例來演示：

工作原理

PS：忘記了該圖出自哪篇文章，畫的挺不錯的，這里直接引用了，如有侵權(quán)，還望聯(lián)系作者。

源碼實現(xiàn)

• observers 訂閱者集合

• _subscribe 添加訂閱者

• next 函數(shù)將所有訂閱者推送相同的數(shù)據(jù)

其他 Subject

操作符（Operator）

由于篇幅問題，本節(jié)并不會細(xì)化到講每個操作符

理解操作符

Operator 本質(zhì)上是一個純函數(shù) (pure function)，它接收一個 Observable 作為輸入，并生成一個新的 Observable 作為輸出。

export interface Operator<T, R> {
  call(subscriber: Subscriber<R>, source: any): TeardownLogic;
}
// 等價于
function Operator(subscriber: Subscriber<R>, source: any){}

遵循的小道

迭代器模式和集合的函數(shù)式編程模式以及管道思想（pipeable）

函數(shù)式編程

操作符的實現(xiàn)以及使用均依照函數(shù)式的編程范式，F(xiàn)unctional Programing，簡稱 FP，函數(shù)式編程范式，它的思維就是一切用函數(shù)表達(dá)和解決問題，避免用命令式。

優(yōu)點：

• 鏈?zhǔn)秸{(diào)用/組合開發(fā)

• 簡單易寫易讀（聲明式）

• 可靠性（純函數(shù)不存在依賴）

• 惰性求值（高階函數(shù)）

• 易于測試

更多詳細(xì)看這篇 不完全指南

pipe

管道，用來承載數(shù)據(jù)流的容器，相信大家一定用過 Lodash 的chain，原生 js 數(shù)組，NodeJS 開發(fā)者 也許還知道 async/bluebird 的 waterfall，Mongodb 的 pipe，它們都遵循管道思想，最直接的好處是鏈?zhǔn)秸{(diào)用，還可以用來劃分邏輯，在異步的場景中還可以做流程控制（串行、并行、競速等等）。

為什么要有操作符？

遵循符合響應(yīng)式宣言，單向線性的通訊或傳輸數(shù)據(jù)，pipe 可以降低耦合度以便于閱讀和維護(hù)，把復(fù)雜的問題分解成多個簡單的問題，最后在組合起來。

操作符與數(shù)據(jù)流

在 RxJS 的世界解決問題的方式是抽象為數(shù)據(jù)流，整個閉環(huán)是圍繞數(shù)據(jù)流進(jìn)行的，所以我們再來理解一下數(shù)據(jù)流：流，可以把數(shù)據(jù)可以想像成現(xiàn)實中的水流，河流，流有上游、下游每個階段處理不同的事情，在這過程避免不了要操作流，比如合并、流程控制、頻率控制等等，所以操作符就扮演了此角色。

生命周期：創(chuàng)建流（create、new、創(chuàng)建類操作符）——> 執(zhí)行流（subscribe） ——> 銷毀流（unsubscribe）

分類

工作原理

迭代器模式：當(dāng)多個操作符時，組合成多個可迭代對象的集合，執(zhí)行時依次調(diào)用 next 函數(shù)。

源碼實現(xiàn)

1. 操作符傳入 pipe

2. pipe 將操作符轉(zhuǎn)換成可迭代的 Array

3. subscribe（執(zhí)行流）時消費操作符邏輯

如圖

操作符轉(zhuǎn)換 Array 源碼

export function pipeFromArray(fns: Array<Function>): Function {
    if (fns.length === 0) {
        return (x: any) => x;
    }
    if (fns.length === 1) {
        return fns[0];
    }
    return (input: any) => {
        return fns.reduce((prev: any, fn: Function) => fn(prev), input);
    };
}

創(chuàng)建自定義操作符

方式一

const isEven = () => {
    return (source: Observable<any>) => {
        return new Observable<any>(observer => {
            const subscription = source.subscribe((x) => {
                observer.next(x % 2 === 0);
                observer.complete();
            })
            return () => subscription.unsubscribe();
        })
    }
}
new Observable(observer => {
    observer.next(7);
})
    .pipe(isEven())
    .subscribe(console.log);
// 執(zhí)行結(jié)果：false

方式二：基于 lift

const odd = () => {
    const operator: Operator<any, any> = {
        call(subscriber: Subscriber<any>, source: any) {
            const subscription = source.subscribe((x: any) => subscriber.next(x % 2 !== 0));
            return () => {
                subscription.unsubscribe();
            };
        },
    }
    return operator;
}
new Observable(observer => {
    observer.next(7);
})
    .lift(odd())
    .subscribe(console.log)
// 執(zhí)行結(jié)果 true

lift 源碼

閱讀彈珠/大理石圖

學(xué)會閱讀彈珠圖是快速理解 Rx 操作符的手段之一，有些操作符需要描述時間流逝以及序列，所以彈珠圖有很多的標(biāo)識和符號，如下圖。

這里有幾個用來理解大理石圖的網(wǎng)站：

• https://rxviz.com/

• https://rxmarbles.com/

學(xué)習(xí)參考

• Async.js

• Lodash

調(diào)度器（Scheduler）

何為調(diào)度器

也許你在使用操作符的過程中從未在意過它，但它在 Rx 起著至關(guān)重要的作用，在異步中如何調(diào)度異步任務(wù)是很復(fù)雜的事情（尤其是以線程為核心處理異步任務(wù)的語言），很慶幸的是我們用使用的 JS ，所以不需要過多的關(guān)注線程問題，更友好的是大多數(shù)操作符默認(rèn)幫開發(fā)者選中了合適的調(diào)度模式（下文會講到），以至于我們從忽略了它，但無論如何我們都應(yīng)該對調(diào)度器有基本的了解。

調(diào)度器，Scheduler 用來控制數(shù)據(jù)推送節(jié)奏的，RxJS 有自己的基準(zhǔn)時鐘和一套的執(zhí)行規(guī)則，來安排多個任務(wù)/數(shù)據(jù)該如何執(zhí)行。

官方定義：

• Scheduler 是一種數(shù)據(jù)結(jié)構(gòu)

• Scheduler 是一個執(zhí)行環(huán)境

• Scheduler 是一個虛擬時鐘

種類/模式

示例

下面我們舉例略窺一下各個模式的表現(xiàn)。

null/undefined/sync

import { asapScheduler, asyncScheduler, from } from 'rxjs';
function syncSchedulerMain() {
    console.log('before');
    from([1, 2, 3]).subscribe(console.log)
    console.log('after');
}
syncSchedulerMain();
// 執(zhí)行結(jié)果：
// before
// 1
// 2
// 3
// after

asap

function asyncSchedulerMain() {
    console.log('asyncScheduler: before');
    from([1, 2], asyncScheduler).subscribe(console.log)
    Promise.resolve('asyncScheduler: promise').then(console.log);
    console.log('asyncScheduler: after');
}
// 執(zhí)行結(jié)果：
// asapScheduler: before
// asapScheduler: after
// 1
// 2
// asapScheduler: promise

從結(jié)果示，from 的數(shù)據(jù)的輸出順序是在 console.log（同步代碼）之后，promise.then 之前的。

async

function asapSchedulerMain() {
    console.log('asapScheduler: before');
    from([1, 2, 3], asapScheduler).subscribe(console.log)
    Promise.resolve('asapScheduler: promise').then(console.log);
    console.log('asapScheduler: after');
}
// 執(zhí)行結(jié)果：
// asyncScheduler: before
// asyncScheduler: after
// asyncScheduler: promise
// 1
// 2

結(jié)果示，from 數(shù)據(jù)輸出順序是在 console.log（同步代碼）和 Promise.then 之后的。

工作原理

Scheduler 工作原理可以類比 JS 中的調(diào)用棧和事件循環(huán)，從實現(xiàn)上 aspa和 async也的確交給事件循環(huán)來處理。null /undefined相當(dāng)于調(diào)用棧，aspa相當(dāng)于事件循環(huán)中的微任務(wù)，async相當(dāng)于宏任務(wù)，可以肯定的是微任務(wù)執(zhí)行時機的優(yōu)先級比宏任務(wù)要高，所以從執(zhí)行時機來看 null > aspa > async。queue運行模式根據(jù) delay 的參數(shù)來決定，如果是 0，那么就用同步的方式執(zhí)行，如果大于 0，就以 async 模式執(zhí)行。

使用原則/策略

RxJS Scheduler 的原則是：盡量減少并發(fā)運行。

1. 對于返回有限和少量消息的 observable 的操作符，RxJS 不使用調(diào)度器，即 null 或 undefined 。

2. 對于返回潛在大量的或無限數(shù)量的消息的操作符，使用 queue 調(diào)度器。

3. 對于使用定時器的操作符，使用 aysnc 調(diào)度器。

支持調(diào)度器的操作符

of 、 from 、 timer 、 interval 、 concat 、 merge 、 combineLatest ，更多戳 這里。

bufferTime 、 debounceTime 、 delay 、 auditTime 、 sampleTime 、 throttleTime 、 timeInterval 、 timeout 、 timeoutWith 、 windowTime 這樣時間相關(guān)的操作符全部接收調(diào)度器作為最后的參數(shù)，并且默認(rèn)的操作是在 Scheduler.async 調(diào)度器上。

OK，關(guān)于調(diào)度器我們先了解到這里。

最后

至此，RxJS 內(nèi)容已經(jīng)講解完畢，文中概念較多，若大家都能夠理解，就可以對 RxJS 的認(rèn)知拉到同一個維度，后續(xù)需要做的就是玩轉(zhuǎn)各種操作符，解決實際問題，學(xué)以致用才可達(dá)到真正的精通。

最后如果覺得文章不錯，點個贊再走吧！

附文中完整代碼與示例： https://github.com/aaaaaajie/simple-rxjs

推薦閱讀

• 玩轉(zhuǎn) RxJS 操作符 ——流程控制篇

• 玩轉(zhuǎn) RxJS 操作符——回壓控制篇

參考

• 反應(yīng)式宣言

• RxJS 中文文檔

• Reactive X 文檔

• RxJS 入門指南

• RxJS 給你絲滑般的編程體驗

• Observable vs Subject

• 《RxJS 深入淺出》——程墨