RxJS——給你如絲通常順滑的編程體驗（篇幅較長，建議收藏）

前言

懷着對於RxJS這項技術的好奇，筆者花了數天時間研究了這項技術，並肝了一包枸杞才完成這篇文章的撰寫，屬實不易。不過也正是經過這段時間的學習，我發現這項技術在必定程度上能夠解決我在平常業務中遇到的一些痛點，以及有種想立刻應用到本身的新項目中的慾望，的確這種以數據流的理念來管控大型項目中的數據能給人帶來一種十分優雅的編程體驗。html

若是看完以爲有收穫，但願給筆者一個贊呀😘前端

概念

RxJS 是 Reactive Extensions for JavaScript 的縮寫，起源於 Reactive Extensions，是一個基於可觀測數據流 Stream 結合觀察者模式和迭代器模式的一種異步編程的應用庫。RxJS 是 Reactive Extensions 在 JavaScript 上的實現。git

注意！它跟React沒啥關係，筆者最初眼花把它當作了React.js的縮寫（恥辱啊！！！）github

對於陌生的技術而言，咱們通常的思路莫過於，打開百度（google），搜索，而後查看官方文檔，或者從零散的博客當中，去找尋可以理解這項技術的信息。但在不少時候，僅從一些隻言片語中，的確也很難真正瞭解到一門技術的前因後果。web

本文將從學習的角度來解析這項技術具有的價值以及能給咱們現有項目中帶來的好處。面試

背景

從開發者角度來看，對於任何一項技術而言，咱們常常會去談論的，莫過於如下幾點：編程

應用場景？
如何落地？
上手難易程度如何？
爲何須要它？它解決了什麼問題？

針對以上問題，咱們能夠由淺入深的來刨析一下RxJS的相關理念。設計模式

應用場景？

假設咱們有這樣一個需求：數組

咱們上傳一個大文件以後，須要實時監聽他的進度，而且待進度進行到100的時候中止監聽。promise

對於通常的作法咱們能夠採用短輪詢的方式來實現，在對於異步請求的封裝的時候，若是咱們採用Promise的方式，那麼咱們通常的作法就能夠採用編寫一個用於輪詢的方法，獲取返回值進行處理，若是進度沒有完成則延遲必定時間再次調用該方法，同時在出現錯誤的時候須要捕獲錯誤並處理。

顯然，這樣的處理方式無疑在必定程度上給開發者帶來了必定開發和維護成本，由於這個過程更像是咱們在觀察一個事件，這個事件會屢次觸發並讓我感知到，不只如此還要具有取消訂閱的能力，Promise在處理這種事情時的方式其實並不友好，而RxJS對於異步數據流的管理就更加符合這種範式。

引用尤大的話：

我我的傾向於在適合 Rx 的地方用 Rx，可是不強求 Rx for everything。比較合適的例子就是好比多個服務端實時消息流，經過 Rx 進行高階處理，最後到 view 層就是很清晰的一個 Observable，可是 view 層自己處理用戶事件依然能夠沿用現有的範式。

如何落地？

針對現有項目來講，如何與實際結合並保證原有項目的穩定性也的確是咱們應該優先考慮的問題，畢竟任何一項技術若是沒法落地實踐，那麼必然給咱們帶來的收益是比較有限的。

這裏若是你是一名使用Angular的開發者，或許你應該知道Angular中深度集成了Rxjs，只要你使用Angular框架，你就不可避免的會接觸到RxJs相關的知識。

在一些須要對事件進行更爲精確控制的場景下，好比咱們想要監聽點擊事件(click event)，但點擊三次以後再也不監聽。

那麼這個時候引入RxJS進行功能開發是十分便利而有效的，讓咱們能省去對事件的監聽而且記錄點擊的狀態，以及須要處理取消監聽的一些邏輯上的心理負擔。

你也能夠選擇爲你的大型項目引入RxJS進行數據流的統一管理規範，固然也不要給本不適合RxJS理念的場景強加使用，這樣實際帶來的效果可能並不明顯。

上手難易程度如何？

若是你是一名具有必定開發經驗的JavaScript開發者，那麼幾分鐘或許你就能將RxJS應用到一些簡單的實踐中了。

爲何須要它？它解決了什麼問題？

若是你是一名使用JavaScript的開發者，在面對衆多的事件處理，以及複雜的數據解析轉化時，是否經常容易寫出十分低效的代碼或者是臃腫的判斷以及大量髒邏輯語句？

不只如此，在JavaScript的世界裏，就衆多處理異步事件的場景中來看，「麻煩」兩個字彷佛常常容易被提起，咱們能夠先從JS的異步事件的處理方式發展史中來細細品味RxJS帶來的價值。

回調函數時代（callback）

使用場景：

事件回調
Ajax請求
Node API
setTimeout、setInterval等異步事件回調

在上述場景中，咱們最開始的處理方式就是在函數調用時傳入一個回調函數，在同步或者異步事件完成以後，執行該回調函數。能夠說在大部分簡單場景下，採用回調函數的寫法無疑是很方便的，好比咱們熟知的幾個高階函數：

forEach
map
filter

[1, 2, 3].forEach(function (item, index) {
    console.log(item, index);
})
複製代碼

他們的使用方式只須要咱們傳入一個回調函數便可完成對一組數據的批量處理，很方便也很清晰明瞭。

但在一些複雜業務的處理中，咱們若是仍然秉持不拋棄不放棄的想法頑強的使用回調函數的方式就可能會出現下面的狀況：

fs.readFile('a.txt', 'utf-8', function(err, data) {
    fs.readFile('b.txt', 'utf-8', function(err, data1) {
        fs.readFile('c.txt', 'utf-8', function(err, data2) {
            // ......
        })
    })
})
複製代碼

固然做爲編寫者來講，你可能以爲說這個很清晰啊，沒啥很差的。可是若是再複雜點呢，若是調用的函數都不同呢，若是每個回調裏面的內容都十分複雜呢。短時間內本身可能清楚爲何這麼寫，目的是什麼，可是過了一個月、三個月、一年後，你肯定在衆多業務代碼中你還能找回當初的本心嗎？

你會不會火燒眉毛的查找提交記錄，這是哪一個憨批寫的，跟shit......，臥槽怎麼是我寫的。

這時候，面對衆多開發者苦不堪言的回調地域，終於仍是有人出來造福人類了......

Promise時代

Promise最初是由社區提出（畢竟做爲天天與奇奇怪怪的業務代碼打交道的咱們來講，一直用回調頂不住了啊），後來官方正式在ES6中將其加入語言標準，並進行了統一規範，讓咱們可以原生就能new一個Promise。

就優點而言，Promise帶來了與回調函數不同的編碼方式，它採用鏈式調用，將數據一層一層日後拋，而且可以進行統一的異常捕獲，不像使用回調函數就直接炸了，還得在衆多的代碼中一個個try catch。

話很少說，看碼！

function readData(filePath) {
    return new Promise((resolve, reject) => {
        fs.readFile(filePath, 'utf-8', (err, data) => {
            if (err) reject(err);
            resolve(data);
        })
    });
}

readData('a.txt').then(res => {
    return readData('b.txt');
}).then(res => {
    return readData('c.txt');
}).then(res => {
    return readData('d.txt');
}).catch(err => {
    console.log(err);
})

複製代碼

對比一下，這種寫法會不會就更加符合咱們正常的思惟邏輯了，這種順序下，讓人看上去十分舒暢，也更利於代碼的維護。

優勢：

狀態改變就不會再變，任什麼時候候都能獲得相同的結果
將異步事件的處理流程化，寫法更方便

缺點：

沒法取消
錯誤沒法被try catch（可是能夠使用.catch方式）
當處於pending狀態時沒法得知如今處在什麼階段

雖然Promise的出如今必定程度上提升了咱們處理異步事件的效率，可是在須要與一些同步事件的進行混合處理時每每咱們還須要面臨一些並不太友好的代碼遷移，咱們須要把本來放置在外層的代碼移到Promise的內部才能保證某異步事件完成以後再進行繼續執行。

Generator函數

ES6 新引入了 Generator 函數，能夠經過 yield 關鍵字，把函數的執行流掛起，爲改變執行流程提供了可能，從而爲異步編程提供解決方案。形式上也是一個普通函數，但有幾個顯著的特徵：

function關鍵字與函數名之間有一個星號 "*" （推薦緊挨着function關鍵字）
函數體內使用 yield· 表達式，定義不一樣的內部狀態（能夠有多個yield`）
直接調用 Generator函數並不會執行，也不會返回運行結果，而是返回一個遍歷器對象（Iterator Object）
依次調用遍歷器對象的next方法，遍歷 Generator函數內部的每個狀態

function* read(){
    let a= yield '666';
    console.log(a);
    let b = yield 'ass';
    console.log(b);
    return 2
}
let it = read();
console.log(it.next()); // { value:'666',done:false }
console.log(it.next()); // { value:'ass',done:false }
console.log(it.next()); // { value:2,done:true }
console.log(it.next()); // { value: undefined, done: true }
複製代碼

這種模式的寫法咱們能夠自由的控制函數的執行機制，在須要的時候再讓函數執行，可是對於平常項目中來講，這種寫法也是不夠友好的，沒法給與使用者最直觀的感覺。

async / await

相信在通過許多面試題的洗禮後，你們或多或少應該也知道這玩意其實就是一個語法糖，內部就是把Generator函數與自動執行器co進行告終合，讓咱們能以同步的方式編寫異步代碼，十分暢快。

有一說一，這玩意着實好用，要不是要考慮兼容性，真就想大面積使用這種方式。

再來看看用它編寫的代碼有多快樂：

async readFileData() {
    const data = await Promise.all([
        '異步事件一',
        '異步事件二',
        '異步事件三'
    ]);
    console.log(data);
}

複製代碼

直接把它看成同步方式來寫，徹底不要考慮把一堆代碼複製粘貼的一個其餘異步函數內部，屬實簡潔明瞭。

RxJS

它在使用方式上，跟Promise有點像，但在能力上比Promise強大多了，不只僅可以以流的形式對數據進行控制，還內置許許多多的內置工具方法讓咱們能十分方便的處理各類數據層面的操做，讓咱們的代碼如絲通常順滑。

優點：

代碼量的大幅度減小
代碼可讀性的提升
很好的處理異步
事件管理、調度引擎
十分豐富的操做符
聲明式的編程風格

function readData(filePath) {
    return new Observable((observer) => {
        fs.readFile(filePath, 'utf-8', (err, data) => {
            if (err) observer.error(err);
            observer.next(data);
        })
    });
}

Rx.Observable
.forkJoin(readData('a.txt'), readData('b.txt'), readData('c.txt'))
.subscribe(data => console.log(data));

複製代碼

這裏展現的僅僅是RxJS能表達能量的冰山一角，對於這種場景的處理辦法還有多種方式。RxJS 擅長處理異步數據流，並且具備豐富的庫函數。對於RxJS而言，他能將任意的Dom事件，或者是Promise轉換成observables。

前置知識點

在正式進入RxJS的世界以前，咱們首先須要明確和了解幾個概念：

響應式編程（Reactive Programming）
流（Stream）
觀察者模式
迭代器模式

響應式編程（Reactive Programming）

響應式編程（Reactive Programming），它是一種基於事件的模型。在上面的異步編程模式中，咱們描述了兩種得到上一個任務執行結果的方式，一個就是主動輪訓，咱們把它稱爲 Proactive 方式。另外一個就是被動接收反饋，咱們稱爲 Reactive。簡單來講，在 Reactive 方式中，上一個任務的結果的反饋就是一個事件，這個事件的到來將會觸發下一個任務的執行。

響應式編程的思路大概以下：你能夠用包括 Click 和 Hover 事件在內的任何東西建立 Data stream（也稱「流」，後續章節詳述）。Stream 廉價且常見，任何東西均可以是一個 Stream：變量、用戶輸入、屬性、Cache、數據結構等等。舉個例子，想像一下你的 Twitter feed 就像是 Click events 那樣的 Data stream，你能夠監聽它並相應的做出響應。

結合實際，若是你使用過Vue，必然可以第一時間想到，Vue的設計理念不也是一種響應式編程範式麼，咱們在編寫代碼的過程當中，只須要關注數據的變化，沒必要手動去操做視圖改變，這種Dom層的修改將隨着相關數據的改變而自動改變並從新渲染。

流（`Stream`）

流做爲概念應該是語言無關的。文件IO流，Unix系統標準輸入輸出流，標準錯誤流(stdin, stdout, stderr)，還有一開始提到的 TCP 流，還有一些 Web 後臺技術（如Nodejs）對HTTP請求/響應流的抽象，均可以見到流的概念。

做爲響應式編程的核心，流的本質是一個按時間順序排列的進行中事件的序列集合。

對於一流或多個流來講，咱們能夠對他們進行轉化，合併等操做，生成一個新的流，在這個過程當中，流是不可改變的，也就是隻會在原來的基礎返回一個新的stream。

觀察者模式

在衆多設計模式中，觀察者模式能夠說是在不少場景下都有着比較明顯的做用。

觀察者模式是一種行爲設計模式，容許你定義一種訂閱機制，可在對象事件發生時通知多個「觀察」該對象的其餘對象。

用實際的例子來理解，就好比你訂了一個銀行卡餘額變化短信通知的服務，那麼這個時候，每次只要你轉帳或者是購買商品在使用這張銀行卡消費以後，銀行的系統就會給你推送一條短信，通知你消費了多少多少錢，這種其實就是一種觀察者模式。

在這個過程當中，銀行卡餘額就是被觀察的對象，而用戶就是觀察者。

優勢：

下降了目標與觀察者之間的耦合關係，二者之間是抽象耦合關係。
符合依賴倒置原則。
目標與觀察者之間創建了一套觸發機制。
支持廣播通訊

不足：

目標與觀察者之間的依賴關係並無徹底解除，並且有可能出現循環引用。
當觀察者對象不少時，通知的發佈會花費不少時間，影響程序的效率。

迭代器模式

迭代器（Iterator）模式又叫遊標（Sursor）模式，在面向對象編程裏，迭代器模式是一種設計模式，是一種最簡單也最多見的設計模式。迭代器模式能夠把迭代的過程從從業務邏輯中分離出來，它可讓用戶透過特定的接口巡訪容器中的每個元素而不用瞭解底層的實現。

const iterable = [1, 2, 3];
const iterator = iterable[Symbol.iterator]();
iterator.next(); // => { value: "1", done: false}
iterator.next(); // => { value: "2", done: false}
iterator.next(); // => { value: "3", done: false}
iterator.next(); // => { value: undefined, done: true}
複製代碼

做爲前端開發者來講，咱們最常遇到的部署了iterator接口的數據結構不乏有：Map、Set、Array、類數組等等，咱們在使用他們的過程當中，均能使用同一個接口訪問每一個元素就是運用了迭代器模式。

Iterator做用：

爲各類數據結構，提供一個統一的、簡便的訪問接口；
使得數據結構的成員可以按某種次序排列；
爲新的遍歷語法 for...of 實現循環遍歷

在許多文章中，有人會喜歡把迭代器和遍歷器混在一塊兒進行概念解析，其實他們表達的含義是一致的，或者能夠說（迭代器等於遍歷器)。

Observable

表示一個概念，這個概念是一個可調用的將來值或事件的集合。它能被多個observer訂閱，每一個訂閱關係相互獨立、互不影響。

舉個栗子：

假設你訂閱了一個博客或者是推送文章的服務號（微信公衆號之類的），以後只要公衆號更新了新的內容，那麼該公衆號就會把新的文章推送給你，在這段關係中，這個公衆號就是一個Observable，用來產生數據的數據源。

相信看完上面的描述，你應該對Observable是個什麼東西有了必定的瞭解了，那麼這就好辦了，下面咱們來看看在RxJS中如何建立一個Observable。

const Rx = require('rxjs/Rx')

const myObservable = Rx.Observable.create(observer => {
  observer.next('foo');
  setTimeout(() => observer.next('bar'), 1000);
});
複製代碼

咱們能夠調用Observable.create方法來建立一個Observable，這個方法接受一個函數做爲參數，這個函數叫作 producer 函數，用來生成 Observable 的值。這個函數的入參是 observer，在函數內部經過調用 observer.next() 即可生成有一系列值的一個 Observable。

咱們先不該理會observer是個什麼東西，從建立一個Observable的方式來看，其實也就是調用一個API的事，十分簡單，這樣一個簡單的Observable對象就建立出來了。

Observer

一個回調函數的集合，它知道如何去監聽由Observable提供的值。Observer在信號流中是一個觀察者（哨兵）的角色，它負責觀察任務執行的狀態並向流中發射信號。

這裏咱們簡單實現一下內部的構造：

const observer = {
	next: function(value) {
		console.log(value);
	},
	error: function(error) {
		console.log(error)
	},
	complete: function() {
		console.log('complete')
	}
}
複製代碼

在RxJS中，Observer是可選的。在next、error 和 complete處理邏輯部分缺失的狀況下，Observable仍然能正常運行，爲包含的特定通知類型的處理邏輯會被自動忽略。

好比咱們能夠這樣定義：

const observer = {
	next: function(value) {
		console.log(value);
	},
	error: function(error) {
		console.log(error)
	}
}
複製代碼

它依舊是能夠正常的運行。

那麼它又是怎麼來配合咱們在實際戰鬥中使用的呢：

const myObservable = Rx.Observable.create((observer) => {
    observer.next('111')
    setTimeout(() => {
        observer.next('777')
    }, 3000)
})

myObservable.subscribe((text) => console.log(text));
複製代碼

這裏直接使用subscribe方法讓一個observer訂閱一個Observable，咱們能夠看看這個subscribe的函數定義來看看怎麼實現訂閱的：

subscribe(next?: (value: T) => void, error?: (error: any) => void, complete?: () => void): Subscription;
複製代碼

源碼是用ts寫的，代碼即文檔，十分清晰，這裏筆者給你們解讀一下，咱們從入參來看，從左至右依次是next、error，complete，且是可選的，咱們能夠本身選擇性的傳入相關回調，從這裏也就印證了咱們上面所說next、error 和 complete處理邏輯部分缺失的狀況下仍能夠正常運行，由於他們都是可選的。

Subscription與Subject

Subscription

Subscription就是表示Observable的執行，能夠被清理。這個對象最經常使用的方法就是unsubscribe方法，它不須要任何參數，只是用來清理由Subscription佔用的資源。同時，它還有add方法能夠使咱們取消多個訂閱。

const myObservable = Rx.Observable.create(observer => {
  observer.next('foo');
  setTimeout(() => observer.next('bar'), 1000);
});
const subscription = myObservable.subscribe(x => console.log(x));
// 稍後：
// 這會取消正在進行中的 Observable 執行
// Observable 執行是經過使用觀察者調用 subscribe 方法啓動的
subscription.unsubscribe();
複製代碼

Subject (主體)

它是一個代理對象，既是一個 Observable 又是一個 Observer，它能夠同時接受 Observable 發射出的數據，也能夠向訂閱了它的 observer 發射數據，同時，Subject 會對內部的 observers 清單進行多播(multicast)

Subjects 是將任意 Observable 執行共享給多個觀察者的惟一方式

這個時候眼尖的讀者會發現，這裏產生了一個新概念——多播。

那麼多播又是什麼呢？
有了多播是否是還有單播？
他們的區別又是什麼呢？

接下來就讓筆者給你們好好分析這兩個概念吧。

單播

普通的Observable 是單播的，那麼什麼是單播呢？

單播的意思是，每一個普通的 Observables 實例都只能被一個觀察者訂閱，當它被其餘觀察者訂閱的時候會產生一個新的實例。也就是普通 Observables 被不一樣的觀察者訂閱的時候，會有多個實例，無論觀察者是從什麼時候開始訂閱，每一個實例都是從頭開始把值發給對應的觀察者。

const Rx = require('rxjs/Rx')

const source = Rx.Observable.interval(1000).take(3);

source.subscribe((value) => console.log('A ' + value))

setTimeout(() => {
    source.subscribe((value) => console.log('B ' + value))
}, 1000)

// A 0
// A 1
// B 0
// A 2
// B 1
// B 2
複製代碼

看到陌生的調用不要慌，後面會進行詳細解析，這裏的source你能夠理解爲就是一個每隔一秒發送一個從0開始遞增整數的Observable就好了，且只會發送三次（take操做符其實也就是限定拿多少個數就不在發送數據了。）。

從這裏咱們能夠看出兩個不一樣觀察者訂閱了同一個源（source），一個是直接訂閱，另外一個延時一秒以後再訂閱。

從打印的結果來看，A從0開始每隔一秒打印一個遞增的數，而B延時了一秒，而後再從0開始打印，因而可知，A與B的執行是徹底分開的，也就是每次訂閱都建立了一個新的實例。

在許多場景下，咱們可能會但願B可以不從最初始開始接受數據，而是接受在訂閱的那一刻開始接受當前正在發送的數據，這就須要用到多播能力了。

多播

那麼若是實現多播能力呢，也就是實現咱們不論何時訂閱只會接收到實時的數據的功能。

可能這個時候會有小夥伴跳出來了，直接給箇中間人來訂閱這個源，而後將數據轉發給A和B不就好了？

const source = Rx.Observable.interval(1000).take(3);

const subject = {
	observers: [],
	subscribe(target) {
		this.observers.push(target);
	},
	next: function(value) {
		this.observers.forEach((next) => next(value))
	}
}

source.subscribe(subject);

subject.subscribe((value) => console.log('A ' + value))

setTimeout(() => {
	subject.subscribe((value) => console.log('B ' + value))
}, 1000)

// A 0
// A 1
// B 1
// A 2
// B 2

複製代碼

先分析一下代碼，A和B的訂閱和單播裏代碼並沒有差異，惟一變化的是他們訂閱的對象由source變成了subject，而後再看看這個subject包含了什麼，這裏作了一些簡化，移除了error、complete這樣的處理函數，只保留了next，而後內部含有一個observers數組，這裏包含了全部的訂閱者，暴露一個subscribe用於觀察者對其進行訂閱。

在使用過程當中，讓這個中間商subject來訂閱source，這樣便作到了統一管理，以及保證數據的實時性，由於本質上對於source來講只有一個訂閱者。

這裏主要是方便理解，簡易實現了RxJS中的Subject的實例，這裏的中間人能夠直接換成RxJS的Subject類實例，效果是同樣的

const source = Rx.Observable.interval(1000).take(3);

const subject = new Rx.Subject();

source.subscribe(subject);

subject.subscribe((value) => console.log('A ' + value))

setTimeout(() => {
	subject.subscribe((value) => console.log('B ' + value))
}, 1000)

複製代碼

一樣先來看看打印的結果是否符合預期，首先A的打印結果並沒有變化，B首次打印的數字如今是從1開始了，也就當前正在傳輸的數據，這下知足了咱們須要獲取實時數據的需求了。

不一樣於單播訂閱者老是須要從頭開始獲取數據，多播模式可以保證數據的實時性。

除了以上這些，RxJS還提供了Subject的三個變體：

BehaviorSubject
ReplaySubject
AsyncSubject

BehaviorSubject

BehaviorSubject 是一種在有新的訂閱時會額外發出最近一次發出的值的Subject。

]

一樣咱們結合現實場景來進行理解，假設有咱們須要使用它來維護一個狀態，在它變化以後給全部從新訂閱的人都能發送一個當前狀態的數據，這就比如咱們要實現一個計算屬性，咱們只關心該計算屬性最終的狀態，而不關心過程當中變化的數，那麼又該怎麼處理呢？

咱們知道普通的Subject只會在當前有新數據的時候發送當前的數據，而發送完畢以後就不會再發送已發送過的數據，那麼這個時候咱們就能夠引入BehaviorSubject來進行終態維護了，由於訂閱了該對象的觀察者在訂閱的同時可以收到該對象發送的最近一次的值，這樣就能知足咱們上述的需求了。

而後再結合代碼來分析這種Subject應用的場景：

const subject = new Rx.Subject();

subject.subscribe((value) => console.log('A：' + value))

subject.next(1);
// A：1
subject.next(2);
// A：2

setTimeout(() => {
	subject.subscribe((value) => console.log('B：' + value)); // 1s後訂閱，沒法收到值
}, 1000)
複製代碼

首先演示的是採用普通Subject來做爲訂閱的對象，而後觀察者A在實例對象subject調用next發送新的值以前訂閱的，而後觀察者是延時一秒以後訂閱的，因此A接受數據正常，那麼這個時候因爲B在數據發送的時候還沒訂閱，因此它並無收到數據。

那麼咱們再來看看採用BehaviorSubject實現的效果：

const subject = new Rx.BehaviorSubject(0); // 須要傳入初始值

subject.subscribe((value: number) => console.log('A：' + value))
// A：0
subject.next(1);
// A：1
subject.next(2);
// A：2

setTimeout(() => {
	subject.subscribe((value: number) => console.log('B：' + value))
	// B：2
}, 1000)
複製代碼

一樣從打印的結果來看，與普通Subject的區別在於，在訂閱的同時源對象就發送了最近一次改變的值（若是沒改變則發送初始值），這個時候咱們的B也如願獲取到了最新的狀態。

這裏在實例化BehaviorSubject的時候須要傳入一個初始值。

ReplaySubject

在理解了BehaviorSubject以後再來理解ReplaySubject就比較輕鬆了，ReplaySubject會保存全部值，而後回放給新的訂閱者，同時它提供了入參用於控制重放值的數量（默認重放全部）。

]

什麼？還不理解？看碼：

const subject = new Rx.ReplaySubject(2);

subject.next(0);
subject.next(1);
subject.next(2);

subject.subscribe((value: number) => console.log('A：' + value))
// A：1
// A：2

subject.next(3);
// A：3
subject.next(4);
// A：4

setTimeout(() => {
	subject.subscribe((value: number) => console.log('B：' + value))
	// B：3
	// B：4
}, 1000)

// 總體打印順序：
// A：1
// A：2
// A：3
// A：4
// B：3
// B：4
複製代碼

咱們先從構造函數傳參來看，BehaviorSubject與ReplaySubject都須要傳入一個參數，對BehaviorSubject來講是初始值，而對於ReplaySubject來講就是重放先前多少次的值，若是不傳入重放次數，那麼它將重放全部發射過的值。

從結果上看，若是你不傳入肯定的重放次數，那麼實現的效果與以前介紹的單播效果幾乎沒有差異。

因此咱們再分析代碼能夠知道在訂閱的那一刻，觀察者們就能收到源對象前多少次發送的值。

AsyncSubject

AsyncSubject 只有當 Observable 執行完成時(執行complete())，它纔會將執行的最後一個值發送給觀察者，若是因異常而終止，AsyncSubject將不會釋聽任何數據，可是會向Observer傳遞一個異常通知。

]

AsyncSubject通常用的比較少，更多的仍是使用前面三種。

const subject = new Rx.AsyncSubject();
subject.next(1);
subject.subscribe(res => {
	console.log('A:' + res);
});
subject.next(2);
subject.subscribe(res => {
	console.log('B:' + res);
});
subject.next(3);
subject.subscribe(res => {
	console.log('C:' + res);
});
subject.complete();
subject.next(4);

// 總體打印結果：
// A:3
// B:3
// C:3
複製代碼

從打印結果來看其實已經很好理解了，也就是說對於全部的觀察者們來講，源對象只會在全部數據發送完畢也就是調用complete方法以後纔會把最後一個數據返回給觀察者們。

這就比如小說裏常常有的，當你要放技能的時候，先要打一套起手式，打完以後纔會放出你的大招。

Cold-Observables與Hot-Observables

Cold Observables

Cold Observables 只有被 observers 訂閱的時候，纔會開始產生值。是單播的，有多少個訂閱就會生成多少個訂閱實例，每一個訂閱都是從第一個產生的值開始接收值，因此每一個訂閱接收到的值都是同樣的。

若是你們想要參考Cold Observables相關代碼，直接看前面的單播示例就好了。

正如單播描述的能力，無論觀察者們何時開始訂閱，源對象都會從初始值開始把全部的數都發給該觀察者。

Hot Observables

Hot Observables 無論有沒有被訂閱都會產生值。是多播的，多個訂閱共享同一個實例，是從訂閱開始接受到值，每一個訂閱接收到的值是不一樣的，取決於它們是從何時開始訂閱。

這裏有幾種場景，咱們能夠逐一分析一下便於理解：

「加熱」

首先能夠忽略代碼中出現的陌生的函數，後面會細說。

const source = Rx.Observable.of(1, 2).publish();
source.connect();
source.subscribe((value) => console.log('A：' + value));
setTimeout(() => {
	source.subscribe((value) => console.log('B：' + value));
}, 1000);
複製代碼

這裏首先用Rx的操做符of建立了一個Observable，而且後面跟上了一個publish函數，在建立完以後調用connect函數進行開始數據發送。

最終代碼的執行結果就是沒有任何數據打印出來，分析一下緣由其實也比較好理解，因爲開啓數據發送的時候尚未訂閱，而且這是一個Hot Observables，它是不會理會你是否有沒有訂閱它，開啓以後就會直接發送數據，因此A和B都沒有接收到數據。

固然你這裏若是把connect方法放到最後，那麼最終的結果就是A接收到了，B仍是接不到，由於A在開啓發數據以前就訂閱了，而B還要等一秒。

更直觀的場景

正如上述多播所描述的，其實咱們更多想看到的現象是可以A和B兩個觀察者可以都有接收到數據，而後觀察數據的差異，這樣會方便理解。

這裏直接換一個發射源：

const source = Rx.Observable.interval(1000).take(3).publish();
source.subscribe((value: number) => console.log('A：' + value));
setTimeout(() => {
	source.subscribe((value: number) => console.log('B：' + value));
}, 3000);
source.connect();

// A：0
// A：1
// A：2
// B：2
複製代碼

這裏咱們利用interval配合take操做符每秒發射一個遞增的數，最多三個，而後這個時候的打印結果就更清晰了，A正常接收到了三個數，B三秒以後才訂閱，因此只接收到了最後一個數2，這種方式就是上述多播所描述的並沒有一二。

二者對比

Cold Observables：舉個栗子會比較好理解一點：好比咱們上B站看番，更新了新番，咱們不論何時去看，都能從頭開始看到完整的劇集，與其餘人看不看毫無關聯，互不干擾。
Hot Observables：這就比如咱們上B站看直播，直播開始以後就直接開始播放了，無論是否有沒有訂閱者，也就是說若是你沒有一開始就訂閱它，那麼你過一段時候後再去看，是不知道前面直播的內容的。

上述代碼中出現的操做符解析

在建立Hot Observables時咱們用到了publish與connect函數的結合，其實調用了publish操做符以後返回的結果是一個ConnectableObservable，而後該對象上提供了connect方法讓咱們控制發送數據的時間。

publish：這個操做符把正常的 Observable（Cold Observables ）轉換成 ConnectableObservable。
ConnectableObservable：ConnectableObservable 是多播的共享 Observable，能夠同時被多個 observers共享訂閱，它是 Hot Observables。ConnectableObservable 是訂閱者和真正的源頭 Observables（上面例子中的 interval，每隔一秒發送一個值，就是源頭 Observables）的中間人，ConnectableObservable 從源頭 Observables 接收到值而後再把值轉發給訂閱者。
connect()：ConnectableObservable 並不會主動發送值，它有個 connect方法，經過調用 connect 方法，能夠啓動共享 ConnectableObservable 發送值。當咱們調用 ConnectableObservable.prototype.connect 方法，無論有沒有被訂閱，都會發送值。訂閱者共享同一個實例，訂閱者接收到的值取決於它們什麼時候開始訂閱。

其實這種手動控制的方式還挺麻煩的，有沒有什麼更加方便的操做方式呢，好比監聽到有訂閱者訂閱了纔開始發送數據，一旦全部訂閱者都取消了，就中止發送數據？其實也是有的，讓咱們看看引用計數（refCount）：

引用計數

這裏主要用到了publish結合refCount實現一個「自動擋」的效果。

const source = Rx.Observable.interval(1000).take(3).publish().refCount();
setTimeout(() => {
	source.subscribe(data => { console.log("A：" + data) });
	setTimeout(() => {
		source.subscribe(data => { console.log("B：" + data) });
	}, 1000);
}, 2000);

// A：0
// A：1
// B：1
// A：2
// B：2
複製代碼

咱們透過結果看本質，可以很輕鬆的發現，只有當A訂閱的時候纔開始發送數據（A拿到的數據是從0開始的），而且當B訂閱時，也是隻能獲取到當前發送的數據，而不能獲取到以前的數據。

不只如此，這種「自動擋」當全部訂閱者都取消訂閱的時候它就會中止再發送數據了。

Schedulers(調度器)

用來控制併發而且是中央集權的調度員，容許咱們在發生計算時進行協調，例如 setTimeout 或 requestAnimationFrame 或其餘。

調度器是一種數據結構。它知道如何根據優先級或其餘標準來存儲任務和將任務進行排序。
調度器是執行上下文。它表示在什麼時候何地執行任務(舉例來講，當即的，或另外一種回調函數機制(好比 setTimeout 或 process.nextTick)，或動畫幀)。
調度器有一個(虛擬的)時鐘。調度器功能經過它的 getter 方法 now() 提供了「時間」的概念。在具體調度器上安排的任務將嚴格遵循該時鐘所表示的時間。

學到這相信你們也已經或多或少對RxJS有必定了解了，不知道你們有沒有發現一個疑問，前面所展現的代碼示例中有同步也有異步，而筆者卻沒有顯示的控制他們的執行，他們的這套執行機制究竟是什麼呢？

其實他們的內部的調度就是靠的Schedulers來控制數據發送的時機，許多操做符會預設不一樣的Scheduler，因此咱們不須要進行特殊處理他們就能良好的進行同步或異步運行。

const source = Rx.Observable.create(function (observer: any) {
    observer.next(1);
    observer.next(2);
    observer.next(3);
    observer.complete();
});

console.log('訂閱前');
source.observeOn(Rx.Scheduler.async) // 設爲 async
.subscribe({
    next: (value) => { console.log(value); },
    error: (err) => { console.log('Error: ' + err); },
    complete: () => { console.log('complete'); }
});
console.log('訂閱後');

// 訂閱前
// 訂閱後
// 1
// 2
// 3
// complete
複製代碼

從打印結果上來看，數據的發送時機的確已經由同步變成了異步，若是不進行調度方式修改，那麼「訂閱後」的打印應該是在數據發送完畢以後纔會執行的。

看完示例以後咱們再來研究這個調度器能作哪幾種調度：

queue
asap
async
animationFrame

queue

將每一個下一個任務放在隊列中，而不是當即執行

queue 延遲使用調度程序時，其行爲與 async 調度程序相同。

當沒有延遲使用時，它將同步安排給定的任務-在安排好任務後當即執行。可是，當遞歸調用時（即在已調度的任務內部），將使用隊列調度程序調度另外一個任務，而不是當即執行，該任務將被放入隊列並等待當前任務完成。

這意味着，當您使用 queue 調度程序執行任務時，您肯定它會在該調度程序調度的其餘任何任務開始以前結束。

這個同步與咱們日常理解的同步可能不太同樣，筆者當時也都困惑了一會。

仍是用一個官方的例子來說解這種調度方式是怎麼理解吧：

import { queueScheduler } from 'rxjs';
queueScheduler.schedule(() => {
  queueScheduler.schedule(() => console.log('second'));

  console.log('first');
});
複製代碼

咱們無需關注陌生的函數調用，咱們這裏着重於看這種調度方式與日常的同步調度的區別。

首先咱們調用queueScheduler的schedule方法開始執行，而後函數內部又一樣再以一樣的方式調用（這裏也能夠改爲遞歸，不過這裏用這個示例去理解可能會好一點），而且傳入一個函數，打印second。

而後繼續看下面的語句，一個普通的console.log('first')，而後咱們再來看看打印結果：

// first
// second
複製代碼

是否是有點神奇，若是沒看明白爲啥的，能夠再回頭看看前面queue對於遞歸執行的處理方式。也就是說若是遞歸調用，它內部會維護一個隊列，而後等待先加入隊列的任務先執行完成（也就是上面的console.log('first')執行完纔會執行console.log('second')，由於console.log('second')這個任務是後加入該隊列的）。

asap

內部基於Promise實現（Node端採用process.nextTick），他會使用可用的最快的異步傳輸機制，若是不支持Promise或process.nextTick或者Web Worker的 MessageChannel也可能會調用setTimeout方式進行調度。

async

與asap方式很像，只不過內部採用setInterval 進行調度，大多用於基於時間的操做符。

animationFrame

從名字看其實相信你們已經就能略知一二了，內部基於requestAnimationFrame來實現調度，因此執行的時機將與window.requestAnimationFrame保持一致，適用於須要頻繁渲染或操做動畫的場景。

Operators

Operator概念

採用函數式編程風格的純函數 (pure function)，使用像 map、filter、concat、flatMap 等這樣的操做符來處理集合。也正由於他的純函數定義，因此咱們能夠知道調用任意的操做符時都不會改變已存在的Observable實例，而是會在原有的基礎上返回一個新的Observable。

儘管 RxJS 的根基是 Observable，但最有用的仍是它的操做符。操做符是容許複雜的異步代碼以聲明式的方式進行輕鬆組合的基礎代碼單元。

實現一個Operator

假設咱們不使用RxJS提供的過濾操做符，那麼讓你本身實現又該怎麼作呢？

function filter(source, callback) {
    return Rx.Observable.create(((observer) => {
        source.subscribe(
            (v) => callback(v) && observer.next(v),
            (err) => observer.error(err),
            (complete) => observer.complete(complete)
        );
    }))
}
const source = Rx.Observable.interval(1000).take(3);
filter(source, (value) => value < 2).subscribe((value) => console.log(value));

// 0
// 1
複製代碼

這樣就實現了一個簡單的filter操做符，是否是很簡潔，其實主要的作法仍是像上面所說，基於傳入的Observable，返回一個新的Observable。

代碼中首先建立了一個Observable，接着用一個新的觀察者訂閱傳入的源，並調用回調函數判斷是否這個值須要繼續下發，若是爲false，則直接跳過，根據咱們傳入的源與過濾函數來看，源對象最終會發送三個數0、一、2，打印結果爲0、1，2被過濾了。

固然咱們也能夠將其放置到Rx.Observable.prototype上以便以咱們能夠採用this的方式獲取源：

Rx.Observable.prototype.filter = function (callback) {
    return Rx.Observable.create(((observer) => {
        this.subscribe(
            (v) => callback(v) && observer.next(v),
            (err) => observer.error(err),
            (complete) => observer.complete(complete)
        );
    }))
}
Rx.Observable.interval(1000).take(3).filter((value) => value < 2).subscribe((value) => console.log(value));

// 0
// 1
複製代碼

這樣是不會就更加簡潔了，就像咱們使用原生數組的filter方法同樣。

要說這兩種方式的區別，其實也比較好理解，一個是放在prototype中，可以被實例化的對象直接調用，另外一個是定義了一個新的函數，能夠用來導出給調用者使用（其實也能夠直接掛載到Observable的靜態屬性上）。

看到這裏估計會有讀者已經猜到筆者接下來講講解什麼了。

實例操做符-靜態操做符

實例操做符：一般是能被實例化的對象直接調用的操做符。咱們通常更多會使用實例操做符多一點，好比filter、map、concat等等。使用實例操做符能夠更快樂的使用this，而省去一個參數，還能維持鏈式調用。
靜態操做符：Observable是一個class類，咱們能夠直接把操做符掛載到他的靜態屬性上，好處在於無需實例化便可調用，缺點在於就沒法再使用this的方式進行目標對象調用了，而是須要把目標對象傳入。

若是添加一個實例化屬性上面已經有示例了，這裏就不作過多贅述了。

將上述的filter例子改造一下，將其掛載到靜態屬性上：

Rx.Observable.filter = (source, callback) => {
    return Rx.Observable.create(((observer) => {
        source.subscribe(
            (v) => callback(v) && observer.next(v),
            (err) => observer.error(err),
            (complete) => observer.complete(complete)
        );
    }))
}
複製代碼

建立型Operators

對於任何數據的處理或使用來講，咱們首先會去關注的莫過於，它從哪裏來，如何產生的，以及咱們該怎麼獲取。

create

定義：

public static create(onSubscription: function(observer: Observer): TeardownLogic): Observable

通過前面代碼的洗禮，相信你們對該操做符已經不陌生了。

create 將 onSubscription 函數轉化爲一個實際的 Observable 。每當有人訂閱該 Observable 的時候，onSubscription函數會接收 Observer 實例做爲惟一參數執行。onSubscription 應該調用觀察者對象的 next, error 和 complete 方法。

官方文檔的描述其實已經很清晰了，至關於只要有人訂閱該操做符建立出來的Observable，它則會經過調用訂閱者自己的方法傳遞一系列值。

上圖與演示代碼並沒有直接關聯。

const source = Rx.Observable.create(((observer: any) => {
    observer.next(1);
    observer.next(2);
    setTimeout(() => {
        observer.next(3);
    }, 1000)
}))

// 方式一
source.subscribe(
    {
        next(val) {
            console.log('A：' + val);
        }
    }
);
// 方式二
source.subscribe((val) => console.log('B：' + val));

// A：1
// A：2
// B：1
// B：2
//- 1s後:
// A：3
// B：3
複製代碼

打印結果天然是不用多提了，首先A和B都會分別打印，一、2，並在1s後打印出3。

這裏咱們能夠注意一下，咱們的在調用subscribe的時候能夠使用這兩種方式，以一個對象形式，該對象具有next、error、complete三個方法（都是可選的），或者直接傳入函數的方式，參數先後分別爲next、error、complete。

empty

定義：

public static empty(scheduler: Scheduler): Observable

顧名思義，該操做符建立一個什麼數據都不發出，直接發出完成通知的操做符。

這裏可能會有讀者問了，那這玩意有啥用。

其實否則，在與某些操做符進行配合時，它的做用還真不可小覷，好比mergeMap，後面會進行配合講解，等不及的小夥伴能夠直接跳到mergeMap。

from

定義：

public static from(ish: ObservableInput<T>, scheduler: Scheduler): Observable<T>

從一個數組、類數組對象、Promise、迭代器對象或者類 Observable 對象建立一個 Observable.

該方法就有點像js中的Array.from方法（能夠從一個類數組或者可迭代對象建立一個新的數組），只不過在RxJS中是轉成一個Observable給使用者使用。

const source = Rx.Observable.from([10, 20, 30]);
source.subscribe(v => console.log(v));

// 10
// 20
// 30
複製代碼

從示例代碼來看，其實這個仍是比較簡單的用法，若是說你想對現有項目的一些數據（好比數組或類數組）採用RxJS來管理，那麼from操做將是一個不錯的選擇。

fromEvent

定義：

public static fromEvent(target: EventTargetLike, eventName: string, options: EventListenerOptions, selector: SelectorMethodSignature<T>): Observable<T>

建立一個 Observable，該 Observable 發出來自給定事件對象的指定類型事件。可用於瀏覽器環境中的Dom事件或Node環境中的EventEmitter事件等。

假設咱們有一個這樣的需求，監聽按鈕點擊事件，並打印出來：

const click = Rx.Observable.fromEvent(document.getElementById('btn'), 'click');
click.subscribe(x => console.log(x));
複製代碼

對比咱們使用addEventListener方式來監聽是否是這種寫法更爲流暢。

fromPromise

定義：

public static fromPromise(promise: PromiseLike<T>, scheduler: Scheduler): Observable<T>

從命名上看其實已經很明顯了，就是將Promise轉換成Observable，這樣咱們在編寫代碼時就能夠不用寫.then、.catch之類的鏈式調用了。

若是 Promise resolves 一個值, 輸出 Observable 發出這個值而後完成。若是 Promise 被 rejected, 輸出 Observable 會發出相應的錯誤。

const source = Rx.Observable.fromPromise(fetch('http://localhost:3000'));
source.subscribe(x => console.log(x), e => console.error(e));
複製代碼

這裏爲了演示效果，本地起了一個服務用於測試，自測的時候能夠用別的。

這樣咱們就能輕鬆拿到該請求的返回值了。

interval

定義：

public static interval(period: number, scheduler: Scheduler): Observable

使用該操做符建立的Observable能夠在指定時間內發出連續的數字，其實就跟咱們使用setInterval這種模式差很少。在咱們須要獲取一段連續的數字時，或者須要定時作一些操做時均可以使用該操做符實現咱們的需求。

const source = Rx.Observable.interval(1000);
source.subscribe(v => console.log(v));

複製代碼

默認從0開始，這裏設定的時間爲1s一次，它會持續不斷的按照指定間隔發出數據，通常咱們能夠結合take操做符進行限制發出的數據量。

of

定義：

public static of(values: ...T, scheduler: Scheduler): Observable<T>

與from的能力差不太多，只不過在使用的時候是傳入一個一個參數來調用的，有點相似於js中的concat方法。一樣也會返回一個Observable，它會依次將你傳入的參數合併並將數據以同步的方式發出。

const source = Rx.Observable.of(1, 2, 3);
source.subscribe(v => console.log(v));

// 1
// 2
// 3
複製代碼

依次打印一、二、3.

repeat

定義：

public repeat(count: number): Observable

將數據源重複n次，n爲你傳入的數字類型參數。

const source = Rx.Observable.of(1, 2, 3).repeat(3);
source.subscribe(v => console.log(v));
複製代碼

這裏配合of操做符，打印結果爲一次打印一、二、三、一、二、三、一、二、3，將本來只會打印一次的一、二、3轉化成三次。

range

定義：

public static range(start: number, count: number, scheduler: Scheduler): Observable

建立一個 Observable ，它發出指定範圍內的數字序列。

學過Python的小夥伴有木有一點似曾相識的感受。

const source = Rx.Observable.range(1, 4);
source.subscribe(v => console.log(v));
複製代碼

打印結果：一、二、三、4。

是否是倍感簡單呢。

轉換操做符

那麼什麼是轉換操做符呢，衆所周知，咱們在平常業務中，老是須要與各類各樣的數據打交道，不少時候咱們都不是直接就會對傳輸過來的數據進行使用，而是會對其作必定的轉換，讓他成爲更加契合咱們需求的形狀，這就是轉換操做符的做用所在了。

buffer

定義：

public buffer(closingNotifier: Observable<any>): Observable<T[]>

將過往的值收集到一個數組中，而且僅當另外一個 Observable 發出通知時才發出此數組。這至關於有一個緩衝區，將數據收集起來，等到一個信號來臨，再釋放出去。

;

改操做符就有點像一個大水壩，一些時候咱們會選擇蓄水，等到特定時候，再由領導下命令打開水壩，讓水流出去。

舉個栗子：

假設咱們有這樣一個需求，咱們有一個接口是專門用於獲取特定數據的，可是呢該接口一次性只返回一個數據，這讓咱們很苦惱，由於產品想讓數據量達到特定值再控制進行操做，也就是他點擊一下某個按鈕，再去將這些數據渲染出來，那該怎麼辦呢？

這個時候就須要咱們的buffer操做符大展身手了：

const btn = document.createElement('button');
btn.innerText = '你點我啊！'
document.body.appendChild(btn);
const click = Rx.Observable.fromEvent(btn, 'click');
const interval = Rx.Observable.interval(1000);
const source = interval.buffer(click);
source.subscribe(x => console.log(x));
複製代碼

這裏咱們直接用interval來演示接口獲取數據，而後再配合buffer進行功能實現。

這裏咱們等四秒以後再點擊一下按鈕，打印出來的值爲：[0, 1, 2, 3]，而後再等8秒，點擊按鈕：[4, 5, 6, 7, 8, 9, 10, 11]。

從現象看，咱們不難看出，咱們已經實現了經過按鈕來控制數據的發送。同時咱們能夠發現另外一個現象，發送出去的數據就直接會在緩衝區中被清空，而後從新收集新的數據。

這其實也不難理解，咱們仍是用水壩來舉例，咱們打開水壩放水一段時間以後，而後關閉它繼續蓄水，那麼我第二次打開水壩放出去的水天然是我新蓄的水。

concatMap

定義：

public concatMap(project: function(value: T, ?index: number): ObservableInput, resultSelector: function(outerValue: T, innerValue: I, outerIndex: number, innerIndex: number): any): Observable

這個操做符仍是有點意思的，咱們先看看官網的描述：

將源值投射爲一個合併到輸出 Observable 的 Observable,以串行的方式等待前一個完成再合併下一個 Observable。

不知道各位讀者是否感覺到了「一絲絲」的很差理解呢，不過等筆者舉個小例子就能輕鬆的搞懂了：

假設你遇到了這樣一個場景，你和女友一塊兒在小吃街逛街，可是呢女友有個很差的毛病，她總喜歡這家買完吃一口而後剩下讓你吃，而後另外一家買一點吃一口而後剩下仍是讓你吃，而你呢每次吃東西也是要時間的，通常會心疼男友的女友就會等你吃完再去買下一家的，這種狀況下，你仍是能吃完再休息會；另外一種狀況呢，女友無論你吃完沒，她繼續買買買，而後你手裏的吃的愈來愈多，你吃的速度徹底趕不上女友買的速度，那這個時候呢就會致使你負重愈來愈大，最後頂不住心態爆炸了。

以上情景包含了concatMap的幾個核心點以及須要注意的地方：

源值發送一個數據，而後你傳入的內部Observable就會開始工做或者是發送數據，訂閱者就能收到數據了，也就是內部的Observable至關於老是要等源對象發送一個數據纔會進行新一輪工做，而且要等本輪工做完成了才能繼續下一輪。
若是本輪工做還未完成又接受到了源對象發送的數據，那麼將會用一個隊列保存，而後等本輪完成當即檢查該隊列裏是否還有，若是有則立馬開啓下一輪。
若是內部Observable的工做時間大於源對象發送的數據的間隔時間，那麼就會致使緩存隊列愈來愈大，最後形成性能問題

其實通俗點理解就是，一個工廠流水線，一個負責發材料的，另外一個負責製做產品的，發材料的就是源對象，製做產品的就是這個內部Observable，這個工廠裏產出的只會是成品也就是製做完成的，因此訂閱者要等這個製做產品的人作完一個才能拿到一個。

若是發材料的速度比製做的人制做一個產品要快就會產生材料堆積，那麼隨着時間推移就會越堆越多，致使工廠裝不下。

藉助代碼理解：

const source = Rx.Observable.interval(3000);
const result = source.concatMap(val => Rx.Observable.interval(1000).take(2));
result.subscribe(x => console.log(x));
複製代碼

首先分析一下代碼結構，咱們先建立了一個每隔三秒發送一個數據的源對象，接着調用實例方法concatMap，並給該方法傳入一個返回Observable對象的函數，最終得到通過concatMap轉化後的Observable對象，並對其進行訂閱。

運行結果爲：首先程序運行的第三秒source會發送第一個數據，而後這時咱們傳入的內部Observable，開始工做，通過兩秒發送兩個遞增的數，接着訂閱函數逐步打印出這兩個數，等待一秒後也就是程序運行的第6秒，source發送第二個數，這個時候重複上述流程。

map

定義：

public map(project: function(value: T, index: number): R, thisArg: any): Observable<R>

若是說你使用js中數組的map方法較多的話，可能這裏基本就不用看了，用法徹底一致。

你只須要傳入一個函數，那麼函數的第一個參數就是數據源的每一個數據，第二個參數就是該數據的索引值，你只須要返回一個計算或者其餘操做以後的返回值便可做爲訂閱者實際獲取到的值。

const source = Rx.Observable.interval(1000).take(3);
const result = source.map(x => x * 2);
result.subscribe(x => console.log(x));
複製代碼

take操做符其實也就是限定拿多少個數就不在發送數據了。

這裏用於演示將每一個數據源的值都乘以2而後發送給訂閱者，因此打印的值分別爲：0、二、4。

mapTo

定義：

public mapTo(value: any): Observable

忽略數據源發送的數據，只發送指定的值（傳參）。

就像是一個你討厭的人讓你幫忙傳話，他說了一大堆表白的話，而後讓你傳給某個妹子，你由於討厭他因此不想幫他，因而跟那個妹子說我喜歡你，最後大家幸福的生活在一塊兒了。

const source = Rx.Observable.interval(1000).take(3);
const result = source.mapTo(666);
result.subscribe(x => console.log(x));
複製代碼

就像這段代碼，數據源發送的是0、一、2，而訂閱者實際收到的是三個666。

mergeMap

定義：

public mergeMap(project: function(value: T, ?index: number): ObservableInput, resultSelector: function(outerValue: T, innerValue: I, outerIndex: number, innerIndex: number): any, concurrent: number): Observable

這個定義看上有點嚇人，不過咱們不要慌，咱們只須要了解他得大多數狀況的用法便可。

這裏你是否還記得前面在empty操做符介紹的部分提到的，筆者留了個坑沒補，就是演示mergeMap與empty是如何進行配合的？這裏就把這個坑填上。

const source = Rx.Observable.interval(1000).take(3);
const result = source.mergeMap(x => x % 2 === 0 ? Rx.Observable.of(x) : Rx.Observable.empty());
result.subscribe(x => console.log(x));
複製代碼

輸入源是一個會發送0、一、2三個數的數據源，咱們調用mergeMap操做符，並傳入一個函數，該函數的功能就是，若是輸入源發送的當前值是偶數則發送給訂閱者，不然就不發送。

這裏面mergeMap主要作了一個整合的能力，咱們能夠將它與map進行對比，咱們能夠發現map的返回值必須是一個數值，而mergeMap返回值是要求是一個Observable，也就是說，咱們能夠返回任意轉換或具有其餘能力的Observable。

pluck

定義：

public pluck(properties: ...string): Observable

用於選擇出每一個數據對象上的指定屬性值。

就好比某個數據源發送的數據是一個對象，對象上面有一個name屬性，而且訂閱者指向知道這個name屬性，那麼就能夠使用該操做符來提取該屬性值給用戶。

const source = Rx.Observable.of({name: '張三'}, {name: '李四'});
const result = source.pluck('name');
result.subscribe(x => console.log(x));

// 張三
// 李四
複製代碼

毫無疑問，這個操做符就是爲了提取屬性來的，至關於咱們使用map操做符來處理一下提取出name再返回給訂閱者。

scan

定義：

public scan(accumulator: function(acc: R, value: T, index: number): R, seed: T | R): Observable<R>

累加器操做符，能夠用來作狀態管理，用處挺多。

就用法來看，咱們能夠參考一下js中數組的reduce函數。

假設咱們如今有一個需求，咱們想要將數據源發送過來的數據累加以後再返回給訂閱者，這又該怎麼作呢？

const source = Rx.Observable.interval(1000).take(4);
const result = source.scan((acc, cur) => acc + cur, 0);
result.subscribe(x => console.log(x));
複製代碼

從代碼上看，數據源發送了四個值：0、一、二、3，而訂閱者每次收到的值將分別是前面已接收到的數與當前數的和也就是：0、一、三、6。

而後再看用法，咱們給scan操做符第一個參數傳入了一個函數，接收兩個值：acc（前一次累加的結果或初始值）、cur（當前值），第二個參數則是計算的初始值。

switchMap

定義：

public switchMap(project: function(value: T, ?index: number): ObservableInput, resultSelector: function(outerValue: T, innerValue: I, outerIndex: number, innerIndex: number): any): Observable

其實也就是switch操做符與map操做符的結合，switch操做符會在組合操做符中講到。

主要做用首先會對多個Observable進行合併，而且具有打斷能力，也就是說合並的這個幾個Observable，某個Observable最早開始發送數據，這個時候訂閱者能正常的接收到它的數據，可是這個時候另外一個Observable也開始發送數據了，那麼第一個Observable發送數據就被打斷了，只會發送後來者發送的數據。

用通俗的話來講就是，有人在說話，忽然你大聲開始說話，人家就被你打斷了，這個時候你們就只能聽到你說話了。

const btn = document.createElement('button');
btn.innerText = '我要發言！'
document.body.appendChild(btn);
const source = Rx.Observable.fromEvent(btn, 'click');
const result = source.switchMap(x => Rx.Observable.interval(1000).take(3));
result.subscribe(x => console.log(x));
複製代碼

代碼實現的功能就是，當某位同窗點擊按鈕，則開始從0開始發送數字，這個時候若是同窗一還沒發送完數據，同窗二再點一下，則同窗一的數據就不會再發了，開始發同窗二的。

假設同窗一點完以後，第二秒同窗二點擊了一下按鈕，則打印結果：0、一、0、一、2，這裏從第二個0開始就是同窗二發送的數據了。

其餘轉換操做符

官網傳送門：轉換操做符

bufferCount
bufferTime
bufferToggle
bufferWhen
concatMapTo
exhaustMap
expand
groupBy
mergeMapTo
mergeScan
pairwise
partition
switchMapTo
window
windowCount
windowTime
windowToggle
windowWhen

過濾操做符

debounceTime

定義：

public debounceTime(dueTime: number, scheduler: Scheduler): Observable

可能對於有過必定js開發經驗的小夥伴應該會知道debounce防抖函數，那麼這個時候會有小夥伴問了，它不會就和debounce差很少吧？沒錯，他的功能與debounce防抖函數差很少，不過仍是有一點差異的。

只有在特定的一段時間通過後而且沒有發出另外一個源值，才從源 Observable 中發出一個值。

也就是說，假設一個數據源每隔一秒發送一個數，而咱們使用了debounceTime操做符，並設置了延時時間，那麼在數據源發送一個新數據以後，若是在延時時間內數據源又發送了一個新數據，這個新的數據就會被先緩存住不會發送，等待發送完數據以後並等待延時時間結束纔會發送給訂閱者，不只如此，在延時時間未到的時候而且已有一個值在緩衝區，這個時候又收到一個新值，那麼緩衝區就會把老的數據拋棄，放入新的，而後從新等待延時時間到達而後將其發送。

const source = Rx.Observable.interval(1000).take(3);
const result = source.debounceTime(2000);
result.subscribe(x => console.log(x));
複製代碼

從代碼來看，咱們不妨猜想一下，最後打印的結果是什麼？

首先咱們建立了一個每秒發送一個數字而且只會發三次的數據源，而後用debounceTime處理了一下，並設置延時時間爲2秒，這個時候咱們觀察打印的數據會發現，程序啓動以後的前三秒沒有數據打印，等到五秒到了以後，打印出一個2，接着就沒有再打印了，這是爲何？

答案是數據源會每秒依次發送三個數0、一、2，因爲咱們設定了延時時間爲2秒，那麼也就是說，咱們在數據發送完成以前都是不可能看到數據的，由於發送源的發送頻率爲1秒，延時時間卻有兩秒，也就是除非發送完，不然不可能知足發送源等待兩秒再發送新數據，每次發完新數據以後要等兩秒以後纔會有打印，因此不論咱們該數據源發送多少個數，最終訂閱者收到的只有最後一個數。

throttleTime

定義：

public throttleTime(duration: number, scheduler: Scheduler): Observable<T>

介紹了防抖怎麼能忘了它的老夥伴節流呢？

該操做符主要能力跟咱們認知的節流函數也是一致的，就是它會控制必定時間內只會發送一個數據，多餘的會直接拋棄掉。惟一和防抖操做符不一致的地方就在於它對於第一個值是不會阻塞的。

const source = Rx.Observable.interval(1000).take(6);
const result = source.throttleTime(2000);
result.subscribe(x => console.log(x));

// 0
// 3
複製代碼

打印結果如上所示，其實效果也很容易解釋，代碼中建立了一個數據源每秒發送一個從0開始遞增的數，總共發送6個也就是0-5，並使用throttleTime設置兩秒，訂閱者接收第一個值時不會被阻塞，而是接收完一個以後的兩秒裏都拿不到值，也就是在第四秒的時候才能拿到3。

distinct

定義：

public distinct(keySelector: function, flushes: Observable): Observable

這個操做符也十分好理解，一句話能夠歸納，使用了該操做符，那麼訂閱者收到的數據就不會有重複的了，也就是它是用來過濾重複數據的。

const source = Rx.Observable.from([1, 2, 3, 2, 4, 3]);
const result = source.distinct();
result.subscribe(x => console.log(x));
複製代碼

最終程序運行結果爲：一、二、三、4，重複的數直接被過濾了。

filter

定義：

public filter(predicate: function(value: T, index: number): boolean, thisArg: any): Observable

這種基本應該沒啥好介紹的了，與咱們理解的數組filter方法並沒有差異，只是用的地方不一致。

const source = Rx.Observable.from([1, 2, 3, 2, 4, 3]);
const result = source.filter(x => x !== 3);
result.subscribe(x => console.log(x));
複製代碼

程序運行結果就是除了3之外的其餘值都被打印出來。

first

定義：

public first(predicate: function(value: T, index: number, source: Observable<T>): boolean, resultSelector: function(value: T, index: number): R, defaultValue: R): Observable<T | R>

只發出由源 Observable 所發出的值中第一個(或第一個知足條件的值)。

這個也和上面差很少，基本看介紹就能懂，這裏就再也不多贅述了。

take

定義：

public take(count: number): Observable<T>

只發出源 Observable 最初發出的的N個值 (N = count)。

這個操做符可謂是在前面出現了不少次了，還挺常見的，用於控制只獲取特定數目的值，跟interval這種會持續發送數據的配合起來就能自主控制要多少個值了。

skip

定義：

public skip(count: Number): Observable

返回一個 Observable，該 Observable 跳過源 Observable 發出的前N個值(N = count)。

舉個栗子來講就是，假設這個數據源發送6個值，你能夠使用skip操做符來跳過前多少個。

const source = Rx.Observable.from([1, 2, 3, 2, 4, 3]);
const result = source.skip(2);
result.subscribe(x => console.log(x));
複製代碼

打印結果爲：三、二、四、3，跳過了前面兩個數。

其餘過濾操做符

官方提供的操做符仍是挺多的，這裏就不一一介紹了，感興趣能夠去官網查看：過濾操做符。

debounce
distinctKey
distinctUntilChanged
distinctUntilKeyChanged
elementAt
ignoreElements
audit
auditTime
last
sample
sampleTime
single
skipLast
skipUntil
skipWhile
takeLast
takeUntil
takeWhile
throttle

組合操做符

concatAll

定義：

public concatAll(): Observable

顧名思義，該操做符有點像咱們js中數組方法concat，用於將多個Observable合成一個，不過它有個注意點在於它是串行的，也就是合併了兩個Observable，那訂閱者在獲取值的時候會先獲取完第一個Observable，以後纔開始接收到後一個Observable的值。

const source1 = Rx.Observable.of(1, 2);
const source2 = source1.map(x => Rx.Observable.interval(1000).take(3));
const result = source2.concatAll();
result.subscribe(x => console.log(x));
複製代碼

根據上面的文字介紹，相信你們對於這段代碼應該也能多少看得懂一些，沒錯，這段代碼的含義就是咱們的數據源發送了兩個數，而且採用map操做符處理完返回了一個新的Observable，這個時候爲了訂閱者可以正常的接收多個Observable，則採用concatAll合併一下，而且最終訂閱者收到的結果依次爲：0、一、二、0、一、2。

mergeAll

定義：

public mergeAll(concurrent: number): Observable

與concatAll幾乎沒太大差異，惟一不一樣的就是它是並行的，也就是合併的多個Observable發送數據時是不分前後的。

combineLatest

定義：

public combineLatest(other: ObservableInput, project: function): Observable

組合多個 Observables 來建立一個 Observable ，該 Observable 的值根據每一個輸入 Observable 的最新值計算得出的。

這個操做符光從簡介來看不太好理解，咱們來結合實例進行講解吧。

const s1 = Rx.Observable.interval(2000).take(3);
const s2 = Rx.Observable.interval(1000).take(5);
const result = s1.combineLatest(s2, (a, b) => a + b);
result.subscribe(x => console.log(x));
複製代碼

打印結果依次是：0、一、二、三、四、五、6。

首先咱們看這個combineLatest的使用方式，它是一個實例操做符，這裏演示的是將s1與s2結合到一塊兒，而且第二個參數須要傳入回調，對結合的值進行處理，因爲咱們這裏只結合了兩個，故只接收a、b兩個參數，該回調函數的返回值即爲訂閱者獲取到的值。

從結果看其實也看不出來啥，主要是這個過程以下：

s2發送一個0，而此時s1未發送值，則咱們傳入的回調不會執行，訂閱者也不會接收到值。
s1發送一個0，而s2最後一次發送的值爲0，故調用回調函數，並把這兩個參數傳入，最終訂閱者收到
s2發送一個1，而s1最後一次發送的爲0，故結果爲1。
s1發送一個1，而s2最後一次發送的值爲1，故結果爲2。
s2發送一個值爲2，而s1最後一次發送的值爲1，故結果爲3。
s2發送一個值爲3，而s1最後一次發送的值爲1，故結果爲4。
...重複上述步驟。

這裏有個注意點，咱們會發現s1、s2在某些時候會同時發送數據，可是這個也會有前後順序的，因此這個時候就看他們誰先定義那麼誰就會先發送，從上面步驟中大家應該也能發現這個現象。

其實也就是結合的多個源之間存在一種依賴關係，也就是兩個源都至少發送了一個值，訂閱者纔會收到消息，等到兩個源都發送完畢，最後纔會發出結束信號。

zip

定義：

public static zip(observables: *): Observable<R>

將多個 Observable 組合以建立一個 Observable，該 Observable 的值是由全部輸入 Observables 的值按順序計算而來的。若是最後一個參數是函數, 這個函數被用來計算最終發出的值.不然, 返回一個順序包含全部輸入值的數組.

通俗點說就是多個源之間會進行順位對齊計算，跟前面的combineLatest有點差異。

話很少說，上碼：

const s1 = Rx.Observable.interval(1000).take(3);
const s2 = Rx.Observable.interval(2000).take(5);
const result = s1.zip(s2, (a, b) => a + b);
result.subscribe(x => console.log(x));
複製代碼

打印結果依次是：0、二、4。

怎麼理解呢，首先咱們記住一句話，多個源之間用來計算的數是順位對齊的，也就是說s1的第一個數對齊s2的第一個數，這種一一對應的計算，最終訂閱者收到的就是將多個對齊的數傳入咱們在調用zip的最後一個回調函數，也就是用來計算完值最終返回給用戶的結果，這是可選的。

等到兩個源中的任意一個源結束了以後，總體就會發出結束信號，由於後續不存在能夠對齊的數了。

startWidth

定義：

public startWith(values: ...T, scheduler: Scheduler): Observable

返回的 Observable 會先發出做爲參數指定的項，而後再發出由源 Observable 所發出的項。

怎麼理解呢，其實很好舉例，好比有一串糖葫蘆，總體都是一個顏色，你以爲很差看，因而你在這串糖葫蘆的前面插了幾個顏色不同的糖葫蘆，這個時候用戶吃的時候就會先吃到你插在最前面的糖葫蘆。

const source = Rx.Observable.interval(1000).take(3);
const result = source.startWith(666)
result.subscribe(x => console.log(x));
複製代碼

打印結果爲：66六、0、一、2。

是否是很好理解呢。

switch

定義：

public switch(): Observable<T>

經過只訂閱最新發出的內部 Observable ，將高階 Observable 轉換成一階 Observable 。

對於該操做符的用法其實前面咱們在介紹switchMap這個轉換操做符時就已經說到了，至關於map+switch=switchMap。

舉個栗子：

const btn = document.createElement('button');
btn.innerText = '我要發言！'
document.body.appendChild(btn);
const source = Rx.Observable.fromEvent(btn, 'click');
const source2 = source.map(x => Rx.Observable.interval(1000).take(3));
const result = source2.switch();
result.subscribe(x => console.log(x));
複製代碼

上述代碼實現的效果與switchMap一致，當用戶點擊按鈕時會開始發送數據，當此次數據發送未完成時，再次點擊按鈕，則會開始一個新的發射數據流程，將原先的發射數據流程直接拋棄。

其餘組合操做符

官網傳送門：組合操做符

combineAll
concat
exhaust
forkJoin
merge
race
withLatestFrom
zipAll

多播操做符

官網傳送門：多播操做符

cache
multicast
publish
publishBehavior
publishLast
publishReplay
share

待完善...

錯誤處理操做符

官網傳送門：錯誤處理操做符

catch
retry
retryWhen

待完善...

工具操做符

官網傳送門：工具操做符

do
delay
delayWhen
dematerialize
finally
let
materialize
observeOn
subscribeOn
timeInterval
timestamp
timeout
timeoutWith
toArray
toPromise

待完善...

條件和布爾操做符

官網傳送門：條件和布爾操做符

defaultIfEmpty
every
find
findIndex
isEmpty

待完善...

數學和聚合操做符

官網傳送門：數學和聚合操做符

count
max
min
reduce

待完善...

總結

整體來講，對於RxJS這種數據流形式來處理咱們平常業務中錯綜複雜的數據是十分有利於維護的，而且在不少複雜的數據運算上來講，RxJS可以給咱們帶來許多提升效率的操做符，同時還給咱們帶來了一種新穎的數據操做理念。

咱們能夠將RxJS比喻作能夠發射事件的一種lodash庫，封裝了不少複雜的操做邏輯，讓咱們在使用過程當中可以以更優雅的方式來進行數據轉換與操做。

參考文章

官方文檔

響應式編程入門指南 - 通俗易懂 RxJS

漁人和Rxjs的故事,此次必定教會你前端必會的Rxjs

使用RxJS來構建響應式應用

完全理解RxJS裏面的Observable 、Observer 、Subject

RxJS——給你如絲通常順滑的編程體驗（篇幅較長，建議收藏）

前言

概念

背景

應用場景？

如何落地？

上手難易程度如何？

爲何須要它？它解決了什麼問題？

回調函數時代（callback）

Promise時代

Generator函數

async / await

RxJS

前置知識點

響應式編程（Reactive Programming）

流（Stream）

觀察者模式

迭代器模式

Observable

Observer

Subscription與Subject

Subscription

Subject (主體)

單播

多播

BehaviorSubject

ReplaySubject

AsyncSubject

Cold-Observables與Hot-Observables

Cold Observables

Hot Observables

「加熱」

更直觀的場景

二者對比

上述代碼中出現的操做符解析

引用計數

Schedulers(調度器)

queue

asap

async

animationFrame

Operators

Operator概念

實現一個Operator

實例操做符-靜態操做符

建立型Operators

create

empty

from

fromEvent

fromPromise

interval

of

repeat

range

轉換操做符

buffer

concatMap

map

mapTo

mergeMap

pluck

scan

switchMap

其餘轉換操做符

過濾操做符

debounceTime

throttleTime

distinct

filter

first

take

skip

其餘過濾操做符

組合操做符

concatAll

mergeAll

combineLatest

zip

startWidth

流（`Stream`）