摘要：仿宋可以把想像成一個(gè)可以發(fā)射事件的庫。在中用來處理異步事件的核心概念包括代表了未來可能會(huì)產(chǎn)生的一系列的值或事件的集合回調(diào)函數(shù)的集合，它知道如何去處理上產(chǎn)生的值或者事件，當(dāng)然也包括異常。

又一年要過去了，回顧2017，rxjs始終是我在項(xiàng)目里使用最頻繁的庫，在我看來，它是一個(gè)非常優(yōu)秀的數(shù)據(jù)處理工具。年初的時(shí)候就計(jì)劃寫點(diǎn)什么，礙于目前公司的項(xiàng)目實(shí)在抽不出時(shí)間，這一拖就到了年底。
臨近新年，總算忙里偷閑，但又不知道從何寫起，于是乎偷了個(gè)懶拿起了官方文檔開始翻譯。以下的文字僅供各位參考，還是強(qiáng)烈建議去看官方文檔。

rxjs是一個(gè)使用觀察者模式來整合異步操作和事件系統(tǒng)的js庫，通過一系列可觀測(cè)的流（observable）將它們串聯(lián)起來。Observable是這個(gè)庫的核心類型，此外還包括諸如Observer，Schedulers，Subjects等類型。還包括一些和數(shù)組方法類似或經(jīng)過演化的操作符，用來協(xié)助處理數(shù)據(jù)。

_可以把 rxjs 想像成一個(gè)可以發(fā)射事件的 lodash 庫。_

響應(yīng)式設(shè)計(jì)結(jié)合了觀察者模式，迭代模式和基于集合的函數(shù)式編程風(fēng)格，從而提供了一種處理事件序列的理想方式。

在rxjs中用來處理異步事件的核心概念包括：

observable: 代表了未來可能會(huì)產(chǎn)生的一系列的值或事件的集合；

observer: 回調(diào)函數(shù)的集合，它知道如何去處理observable上產(chǎn)生的值或者事件，當(dāng)然也包括異常。

subscription: 代表當(dāng)前正在訂閱的observable，主要作用是用來取消訂閱行為。

operators: 純函數(shù)，以函數(shù)式的風(fēng)格通過各種各樣的操作符相互配合對(duì)observable上產(chǎn)生的數(shù)據(jù)集合進(jìn)處理。

subject: 相當(dāng)于一個(gè)事件發(fā)射器，允許將相同的值傳遞給多個(gè)訂閱者。

schedulers: 協(xié)調(diào)observable上數(shù)據(jù)的發(fā)射方式。

在javascript中通常使用以下方式來注冊(cè)事件：

var button = document.querySelector("button");

button.addEventListener("click", () => console.log("Clicked"));

使用rxjs的方式實(shí)現(xiàn)如下：

var button = document.querySelector("button");

Rx.Observable.fromEvent(button, "click")
    .subscribe(() => console.log("Clicked"));

使用不純的函數(shù)時(shí)，函數(shù)可能會(huì)改變外部數(shù)據(jù)的狀態(tài)，比如：

var count = 0;

var button = document.querySelector("button");

button.addEventListener("click", () => console.log(`Clicked ${++count} times`));

當(dāng)使用rxjs時(shí)，必須把這些狀態(tài)隔離出去，比如：

var button = document.querySelect("button");

Rx.Observable.fromEvent(button, "click")
    .scan(count => count + 1, 0)
    .subscribe(count => console.log(`Clicked ${count} times`));

這里的scan操作符的行為和數(shù)組的reduce方法的行為非常類似，它提供一個(gè)初始值給迭代函數(shù)，迭代函數(shù)運(yùn)行后的值將作為下一次迭代的初始值。

rxjs提供了一套非常豐富的操作符來幫助用戶控制數(shù)據(jù)或事件如何在observable中進(jìn)行流動(dòng)。假如我們需要控制用戶在一個(gè)按鈕上每秒最多只能點(diǎn)擊一次

使用普通的javascript代碼:

var count = 0;

var rate = 1000;

var lastClick = Date.now() - rate;

var button = document.querySelector("button");

button.addEventListener("click", () => {
    if(Date.now() - lastClick >= rate) {
        console.log(`Clicked ${++count} timers`);
        lastClick = Date.now();
    }
});

使用rxjs：

var button = button.querySelector("button");

Rx.Observable.fromEvent(button, "click")
    .throttleTime(1000)
    .scan(count => count + 1, 0)
    .subscribe(count => console.log(`Click ${count} times`));

諸如此類對(duì)流進(jìn)行控制的操作符還有：filter, delay, debounceTime, take, takeUntil, distinct, distinctUntilChanged等等。

你可以對(duì)流中的數(shù)據(jù)進(jìn)行轉(zhuǎn)換，當(dāng)我們需要得到每一次鼠標(biāo)點(diǎn)擊時(shí)的x軸坐標(biāo)時(shí)，

使用普通的javascript代碼:

var count = 0;

var rate = 1000;

var lastClick = Date.now() - rate;

var button = document.querySelector("button");

button.addEventListener("click", (event) => {
    if(Date.now() - lastClick >= rate) {
        count += event.clientX;
        console.log(count);
        lastClick = Date.now();
    }
});

使用rxjs：

var button = button.querySelector("button");

Rx.Observable.fromEvent(button, "click")
    .throttleTime(1000)
    .map(event => event.ClientX)
    .scan((count,clientX) => count + clientX, 0)
    .subscribe(count => console.log(count));

諸如此類可以產(chǎn)生新值的操作符還有：pluck，pairwise，sample等等。

可觀測(cè)序列可以推送多個(gè)值，并且它的推送方式是‘懶’的，它滿足下面表格中列出的條件

下面這個(gè)例子中的observable在被訂閱時(shí)可以立即推送出1，2，3這個(gè)三個(gè)值，1秒以后推送第四的值4，然后立即結(jié)束。

var observable = Rx.Observable.create(function(observer) {
    observer.next(1);
    observer.next(2);
    observer.next(3);
    setTimeout(() => {
        observer.next(4);
        observer.complete();
    },1000);
});

為了獲取到這個(gè)observable上的值， 我們需要對(duì)它進(jìn)行訂閱：

var observable = Rx.Observable.create(function(observer) {
    observer.next(1);
    observer.next(2);
    observer.next(3);
    setTimeout(() => {
        observer.next(4);
        observer.complete();
    },1000);
});

console.log("just before subscribe");

observable.subscribe({
    next: x => console.log("got value " + x),
    error: err => console.error("something wrong occurred: " + err),
    complete: () => console.log("done"),
});

console.log("just after subscribe);

這段代碼執(zhí)行后將會(huì)得到如下結(jié)果：

just before subscribe
got value 1
got value 2
got value 3
just after subscribe
got value 4
done

這里使用 pull 和 push 來描述值的生產(chǎn)者和消費(fèi)者之間是如何發(fā)生聯(lián)系的，它們是兩種完全不同的協(xié)議。

在pull的系統(tǒng)中，值的消費(fèi)決定什么時(shí)間從生產(chǎn)者上獲取數(shù)據(jù)，生產(chǎn)者本身并不關(guān)心數(shù)據(jù)什么時(shí)間分發(fā)給消費(fèi)者。

每一個(gè)javascript函數(shù)都可以看作一個(gè) pull 類型的系統(tǒng)。函數(shù)可以產(chǎn)生值，但這是通過調(diào)用者主動(dòng)調(diào)用函數(shù)，函數(shù)運(yùn)行產(chǎn)生出值返回給調(diào)用者的方式進(jìn)行的，所以可以理解為調(diào)用者主動(dòng)去函數(shù)上拉取了值。

ES2015中介紹另外兩種 pull 類型的系統(tǒng)，generator函數(shù) 和 iterator。對(duì)于它們來講，遍歷器對(duì)象的 next 方法可以視作值的消費(fèi)者，通過iterator.next()可以獲取到多個(gè)值。

在push的系統(tǒng)中，生產(chǎn)者決定什么時(shí)候發(fā)送值給消費(fèi)者，消費(fèi)者并知道什么時(shí)候可以接收到值。

ES6中的 promise 就是一個(gè)非常典型的 push 系統(tǒng)，promise 將 resolve 的結(jié)果傳遞給注冊(cè)在它內(nèi)部的回調(diào)函數(shù)，這與普通的函數(shù)有很大的不同，回調(diào)函數(shù)何時(shí)可以接收到數(shù)據(jù)完全取決于 promise 什么時(shí)間向它傳遞數(shù)據(jù)。

rxjs的 Observable 也是一種 push 類型的系統(tǒng)，一個(gè) Observable 可以產(chǎn)生多個(gè)值，然后推送給它的訂閱者。

Function 是一個(gè) ‘懶’ 的求值過程，只有在被調(diào)用時(shí)它才會(huì)同步的返回一個(gè)值給調(diào)用者。

generator 也是一個(gè) ’懶‘ 的求值過種，在遍歷的過程中同步的返回0個(gè)或多個(gè)值給調(diào)用者。

Promise 經(jīng)過運(yùn)算后可能產(chǎn)生一個(gè)值，當(dāng)然也可能產(chǎn)生一個(gè)錯(cuò)誤。

Observable 也是一個(gè)‘懶’的求值過程，當(dāng)它被訂閱后可以同步或者異步的產(chǎn)生出0個(gè)或者無限多個(gè)值給調(diào)用者，這個(gè)過程將一直持續(xù)到訂閱被取消或者流結(jié)束。

很多人認(rèn)為Observable就是一個(gè)事件發(fā)射器或可以產(chǎn)生多個(gè)值的promise，實(shí)際上這種觀點(diǎn)是不正確的。在某些情況下Observable的行為可能與事件發(fā)射器的行為類似，例如使用Subject來發(fā)射值時(shí)，但通常情況下它們的行為與事件發(fā)射器有很大的不同。

_可以把 Observable 想像成一個(gè)不接受參數(shù)的函數(shù)，但是它卻可以產(chǎn)生出很多的值。_

請(qǐng)思考以下代碼：

function foo() {
    console.log("hello");
    return 42;
}

var x = foo.call();

console.log(x);

var y = foo.call();

console.log(y);

運(yùn)行后的輸出：

"hello"
42
"hello"
42

使用rxjs的方式：

var foo = Rx.Observable.create(function(observer) {
    console.log("hello");
    observable.next(42);
});

foo.subscribe(function (x) {
    console.log(x);
});

foo.subscribe(function (y) {
    console.log(y);
});

運(yùn)行后的輸出依然是：

"hello"
42
"hello"
42

首先，函數(shù)和Observable的求值過程都是‘懶’的，如果你不調(diào)用foo函數(shù)，console.log("hello")將不會(huì)執(zhí)行，這對(duì)于Observable也是一樣的，只要不訂閱，console.log("hello")同樣也不會(huì)執(zhí)行。另外，每一次的訂閱（subscribe）和調(diào)用（call）都是互不干擾的，兩次函數(shù)調(diào)用會(huì)觸發(fā)兩次執(zhí)行過程，產(chǎn)生兩次副作用，同樣，兩次訂閱也會(huì)觸發(fā)兩次訂閱過程。對(duì)于事件發(fā)射器來說卻不同，它產(chǎn)生的副作用會(huì)傳遞給每一個(gè)事件接收者，它并不理會(huì)有沒有接收者接收事件，都會(huì)把事件發(fā)射出去，這與Observable有很大的不同。

_訂閱一個(gè)Observable與調(diào)用一個(gè)函數(shù)的過程類似。_

你可能會(huì)認(rèn)為Observable都是異步的，事實(shí)上并不是這樣，我們看一下函數(shù)的調(diào)用過程：

console.log("before");
console.log("foo.call()");
console.log("after");

執(zhí)行后的輸出：

"before"
"hello"
42
"after"

然后以同樣的順序，但是使用Observable：

console.log("before");
foo.subscribe(function(x) {
    console.log(x);
});
console.log("after");

你會(huì)發(fā)現(xiàn)結(jié)果是相同的：

"before"
"hello"
42
"after"

這個(gè)結(jié)果證明了foo這個(gè)Observable完全是同步執(zhí)行的，和調(diào)用一個(gè)函數(shù)沒有什么兩樣。

_Observable傳遞值的方式可以是同步也可以是異步的。_

那么Observable與函數(shù)之間有什么樣的區(qū)別？Observable被訂閱后可以返回多個(gè)值，這是普通的函數(shù)所無法做到的。例如，你無法像下面這樣去實(shí)現(xiàn)一個(gè)函數(shù)：

function foo() {
    console.log("hello");
    return 42;
    return 100; // 這句永遠(yuǎn)不會(huì)執(zhí)行
}

因?yàn)楹瘮?shù)只能返回一個(gè)值，return 100; 會(huì)被解釋器忽略，如果是在typescript中，編譯器會(huì)告訴你這里有一個(gè)錯(cuò)誤。然而這對(duì)于Observable來說卻不是問題：

var foo = Rx.Observable.create(function (observer) {
    console.log("hello");
    observer.next(42);
    observer.next(100); // "返回‘另外一個(gè)值
    observer.next(200); // 同上
});

console.log("before");
foo.subscribe(function(x) {
    console.log(x);
});
console.log("after");

上面的代碼執(zhí)行后會(huì)以同步的方式輸出：

"before"
"Hello"
100
42
200
"after"

當(dāng)然， 你也可以讓它以異步的方式來返回值：

var foo = Rx.Observable.create(function (observer) {
    console.log("hello");
    observer.next(42);
    observer.next(100); // "返回‘另外一個(gè)值
    observer.next(200); // 同上
    setTimeout(() => {
        observer.next(300);
    },1000);
});

console.log("before");
foo.subscribe(function(x) {
    console.log(x);
});
console.log("after");

執(zhí)行后的輸出：

"before"
"Hello"
100
42
200
"after"
300

總結(jié)：

func.call() 表示以同步的方式返回一個(gè)值。

observable.subscribe() 表示以同步或者異步的方式返回一些值。

我們可以使用Rx.Observable.create方法或者那些可以創(chuàng)建 Observable 的操作符來創(chuàng)建 Observable，然后使用一個(gè)觀察者來觀察 Observable，當(dāng) Observable 被訂閱時(shí)它就會(huì)通過 next/error/complete/ 方法來通知觀察者，當(dāng)觀察者放棄觀察行為時(shí)如何處理 Observable 運(yùn)行產(chǎn)生的數(shù)據(jù)。這四個(gè)動(dòng)作都被編碼在一個(gè)可觀察的實(shí)例中，但其中一些動(dòng)作與其它動(dòng)作是相關(guān)的，如觀察和訂閱。

使用Observable時(shí)要關(guān)注的核心問題：

如何創(chuàng)建

被誰訂閱

怎么去執(zhí)行

廢棄時(shí)如何處理

Rx.Observable.create方法實(shí)際是 Observable 類的構(gòu)造函數(shù)的別名，它接受一個(gè)參數(shù)——訂閱函數(shù)。下面這個(gè)示例會(huì)創(chuàng)建一個(gè)observable，它會(huì)每秒發(fā)一個(gè) string 類型的值‘hi’ 給它的觀察者。

var observable = Rx.Observable.create(function subscribe(observer) {
    var id = setInterval(() => {
        observer.next("hi");
    },1000);
});

_可以通過 create 方法來創(chuàng)建 Observable，但實(shí)際情況下我們使用更多的是那些可以創(chuàng)建出 Observable 的操作符，例如：from, interval 等。_

上面代碼中的 subscribe 函數(shù)對(duì)于解釋 Observable 來說是非常重要的一部分，接下來解釋如何去理解它。

上面例子中的observable可以這樣被訂閱：

observable.subscribe(x => console.log(x));

observable變量上有 subscribe 方法，同樣 Observable.create(function subscribe(observer) { ... }) 也有subscribe 方法，這并不是一個(gè)巧合。在庫的實(shí)現(xiàn)中，這兩個(gè)subscribe是不一樣的，但在實(shí)際使用中你可以認(rèn)為它們?cè)诟拍钌鲜窍嗟鹊摹?/p>

這個(gè)示例很好的展示了為什么訂閱行為在多個(gè)訂閱者之間是不會(huì)共享的。當(dāng)我們使用一個(gè)觀察者來訂閱 observable 時(shí)，傳入 create 方法的 subscribe 函數(shù)就會(huì)為這個(gè)觀察者運(yùn)行一次。每次調(diào)用 observable.subscribe 都會(huì)為給定的觀察者觸發(fā)它自己的獨(dú)立設(shè)置。

_訂閱Observable和執(zhí)行一個(gè)函數(shù)很類似，都可以提供一個(gè)回調(diào)函數(shù)來接收將要產(chǎn)生的值。_

這個(gè)過程與諸如 addEventListener/removeEventListener 等事件處理API有很大的區(qū)別。在 observable.subscribe 的過程中，給定的觀察者并沒有注冊(cè)成為 Observable 上的監(jiān)聽者。Observable 甚至不需要維護(hù)一個(gè)觀察者名單。

通過 subscribe 這種簡(jiǎn)單的方式就可以使一個(gè) Observable 開始執(zhí)行，并且將產(chǎn)生的數(shù)據(jù)或事件傳遞給當(dāng)前 Observable 環(huán)境上的觀察者。

Observable.create(function subscribe(observer) { ... })內(nèi)部的代碼代表了 Observable 的執(zhí)行，一個(gè)‘懶’執(zhí)行過程——僅在有觀察者訂閱它的時(shí)候執(zhí)行，當(dāng)它被執(zhí)行后就可以以同步或異步的方式不斷的生產(chǎn)值。

Observable 執(zhí)行后可以產(chǎn)生三種類型的值：

’Next‘：發(fā)送一個(gè)正常的值，比如String， Number 或者 Object 等。

‘Error’：發(fā)送一個(gè)javascript的錯(cuò)誤或者拋出異常。

’Complete‘：停止發(fā)送值。

Next類型的通知是最重要也是最常用的一種，它代表了將要傳遞給觀察者的數(shù)據(jù)。Error 和 Complete 類型的通知在整個(gè)Observable的執(zhí)行過程中只可以發(fā)生一次，并且它們是互斥的，只有一個(gè)可以發(fā)生。

以正則表達(dá)式來表述三者之間的關(guān)系的話，應(yīng)該是：

next*(error|complete)?

_在Observable執(zhí)行的過程中，可以發(fā)生0個(gè)或無限多個(gè)Next類型的通知，但是當(dāng) Error 或 Complete 類型的通知中有一個(gè)發(fā)生后，Observable將不會(huì)再傳遞任何數(shù)據(jù)。_

下面的代碼展示了一個(gè)可以產(chǎn)生三個(gè) Next 類型通知的 Observable，最后發(fā)送一個(gè) Complete 通知：

var observable = Rx.Observable.create(function subscribe(observer) {
    observer.next(1);
    observer.next(2);
    observer.next(3);
    observer.complete();
});

Observable 將嚴(yán)格遵守上述規(guī)則，所以試圖在Observable完成后繼續(xù)傳遞值的行為是無效的。

var observable = Rx.Observable.create(function subscribe(observer) {
    observer.next(1);
    observer.next(2);
    observer.next(3);
    observer.complete();
    observer.next(4); // 不會(huì)傳遞成功
});

如果Observable在執(zhí)行過程中發(fā)生異常，我們可以在subscribe函數(shù)內(nèi)部使用 try/catch 塊來捕獲它：

var observable = Rx.Observable.create(function subscribe(observer) {
    try {
        observer.next(1);
        observer.next(2);
        observer.next(3);
        observer.complete();
    } catch(error) {
        observer.error(error); // 傳遞錯(cuò)誤
    }
});

由于 Observable 可能產(chǎn)生出無限多個(gè)值，但觀察者可能在某時(shí)間點(diǎn)想停止觀察行為，這就需要有一種方法來取消 Observable 的執(zhí)行。由于每個(gè)執(zhí)行只對(duì)應(yīng)一個(gè)觀察者，一旦觀察者完成接收值，它就需要知道如何停止執(zhí)行，以避免浪費(fèi)計(jì)算或內(nèi)存資源。

當(dāng)我們調(diào)用 observable.subscribe 時(shí)，觀察者會(huì)被附加在新創(chuàng)建的可觀察上下文中，同時(shí)這次調(diào)用會(huì)返回一個(gè)對(duì)象，它就是 Subscription 。

var subscription = observable.subscribe(x => console.log(x));

Subscription 代表了正在執(zhí)行的上下文，它擁有一個(gè)可以取消 Observable 執(zhí)行的方法——unsubscribe。調(diào)用這個(gè)方法 subscription.unsubscribe() 就可以取消執(zhí)行。

var observable = Rx.Observable.from([10,20,30]);
var subscription = observable.subscribe(x => console.log(x));
// 一段時(shí)間以后
subscription.unsubscribe();

_當(dāng)你對(duì)一個(gè) Observable 進(jìn)行訂閱時(shí)，就會(huì)獲得一個(gè)代表當(dāng)前執(zhí)行的 Subscription ，只需要調(diào)用它的 unsubscribe 方法就可以取消訂閱。_

當(dāng)我們使用 Observable.create 方法來創(chuàng)建 Observable 時(shí)，必須定義當(dāng)取消訂閱時(shí)如何去釋放執(zhí)行時(shí)的資源，此時(shí)可以返回一個(gè)自定義的 unsubscribe 函數(shù)。

我們可以給之前的例子添加上它的 unsubscribe 方法：

var observable  Rx.Observable.create(function subscribe(observer) {
    var intervalID = setInterval(() => {
        observer.next("hi");
    },1000);

    return function unsubscribe() {
        clearInterval(intervalID);
    }
});

與 observable.unsubscribe 一樣，我們可通過調(diào)用 unsubscribe 方法對(duì)這個(gè)流取消訂閱，這兩個(gè) unsubscribe 在概念上是完全相同的。

事實(shí)上在這個(gè)例子中，假如我們把響應(yīng)式的外殼剝離的話，它完全就是一個(gè)普普通通的javascript函數(shù)：

function subscribe(observer) {
    var intervalID = setInterval(() => {
        observer.next("hi");
    },1000);

    return function unsubscribe() {
        clearInterval(intervalID);
    }
}

var unsubscribe = subscribe({next: (x) => console.log(x)});

盡管如此，我們依然有理由去使用Rx這種響應(yīng)式的編程，通過 Observable，Observer，Subscription，配合各種操作符來實(shí)現(xiàn)更高效安全的數(shù)據(jù)處理。

觀察者實(shí)際就是可觀察序列——Observable的消費(fèi)者。Observer 會(huì)設(shè)置一些回調(diào)函數(shù)來對(duì)應(yīng) Observable 的各個(gè)通知類型（next/error/complete)，從這些通知中接收值，然后對(duì)它們進(jìn)行處理。

這是一個(gè)很典型的 Observer：

var observer = {
    next: x => console.log("Observer got a next value: " + x),
    error: err => console.error("Observer got an error: " + err),
    complete: () => console.log("Observer got a complete notification")
};

我們可以使用它來觀察一個(gè) Observable：

observable.subscribe(observer);

_觀察者就是一個(gè)簡(jiǎn)單對(duì)象，擁有三個(gè)方法，每一個(gè)方法對(duì)應(yīng)處理可觀察序列上的相應(yīng)類型的通知。_

Rxjs中的觀察者允許只實(shí)現(xiàn)這三個(gè)方法中的某幾個(gè)，這對(duì) Observable 的執(zhí)行過程不會(huì)產(chǎn)生影響，僅僅是被忽略方法的對(duì)應(yīng)通知無法得到處理，因?yàn)橛^察者想要處理相應(yīng)通知上的數(shù)據(jù)時(shí)必須實(shí)現(xiàn)對(duì)應(yīng)的方法。

下面這個(gè)示例的 Observer 就沒有實(shí)現(xiàn)處理 complete 方法：

var observer = {
    next: x => console.log("Observer got a next value: " + x),
    error: err => console.error("Observer got an error: " + err),
};

當(dāng)需要訂閱一個(gè) Observable 時(shí)，可以只傳入一個(gè)回調(diào)函數(shù)作為參數(shù)，而不必傳入觀察者對(duì)象：

observable.subscribe(x => console.log("Observer got a next value: " + x));

實(shí)際上在庫的內(nèi)部實(shí)現(xiàn)中會(huì)根據(jù)傳入的參數(shù)為我們創(chuàng)建出一個(gè) Observer 對(duì)象，此時(shí)，傳入的第一個(gè)函數(shù)作為next方法的回調(diào)，第二個(gè)是error回調(diào)，第三個(gè)是complete回調(diào)。也就是說我們可以這樣去訂閱一個(gè) Observable ：

observable.subscribe(
    x => console.log("Observer got a next value: " + x),
    err => console.error("Observer got an error: " + err),
    () => console.log("Observer got a complete notification")
);

Subscription是一個(gè)代表著當(dāng)前正在執(zhí)行的 Observable 的對(duì)象。它有一個(gè)非常重要的方法 unsubscribe ，這個(gè)方法不接受任何參數(shù)，僅僅是用來釋放 Observable 執(zhí)行時(shí)的資源。在舊版本的Rxjs中，它也叫做 Disposable 。

var observable = Rx.Observable.interval(1000);
var subscription = observable.subscribe(x => console.log(x));
// 執(zhí)行一段時(shí)間后

// 下面的操作將會(huì)取消上面已經(jīng)訂閱的 Observable 的執(zhí)行。
subscription.unsubscribe();

_Subscription本質(zhì)上可以只有一個(gè) unsubscribe 方法，用來取消已經(jīng)訂閱的 Observable 的執(zhí)行。_

Subscription 可以被組合在一起，這樣只要調(diào)用一次 unsubscribe 方法就可以將多個(gè) Observable 的執(zhí)行取消掉。我們可以通過 subscription 的 add 方法將它們添加在一起。

var observable1 = Rx.Observable.interval(400);
var observable2 = Rx.Observable.interval(300);

var subscription = observable1.subscribe(x => console.log("first: " + x));

var childSubscription = observable2.subscribe(x => console.log("second: " + x));

subscription.add(childSubscription);

setTimeout(() => {
    // 取消上面的兩次訂閱
    subscription.unsubscribe();
},1000);

執(zhí)行后將輸出：

second:0
first:0
second:1
first:1
second:2

當(dāng)然，可以添加，就可以移除，Subscription同樣擁有一個(gè)remove方法，用來從一組subscription 中移除指定的子 subscription。

Subject 是 Observable 的一種特殊類型，它允許把值同時(shí)傳播給多個(gè) Observer，也就是說它是多播的。相比之下，普通的Observable 是單播的，每一個(gè) Observer 在訂閱 Observable 時(shí)都有其獨(dú)立的執(zhí)行上下文。

_Subject 類似于 Observable ，但它可以把值同時(shí)廣播給多個(gè)觀察者。它也類似于一個(gè)事件發(fā)射器，因此它會(huì)維護(hù)一個(gè)屬于自己的監(jiān)聽者列表。_

每一個(gè) Subject 都是一個(gè) Observable。你可以提供一個(gè)訂閱者來訂閱 Subject，從而在它上面取值。從觀察者的角度來看，并不能區(qū)分出數(shù)據(jù)是從 Subject 上獲取的還是從 Observable 上獲取的。

在 Subject 的內(nèi)部實(shí)現(xiàn)中，并不會(huì)產(chǎn)生新的執(zhí)行上下文來傳遞數(shù)據(jù)，它僅僅是簡(jiǎn)單的將 Observer 注冊(cè)在自己的監(jiān)聽者列表中，這與其它的庫或語言中添加事件的機(jī)制是類似的。

每一個(gè) Subject 都是一個(gè) Observer。每一個(gè) Subject 上都有自己的 next，error，complete方法。當(dāng)需要給 Subject 上添加一個(gè)值時(shí)，只要調(diào)用它的next方法，接下來它就會(huì)把這個(gè)值廣播給注冊(cè)在監(jiān)聽者列表中的多個(gè)監(jiān)聽者。

下面的例子中，我們給 Subject 添加了兩個(gè)觀察者，然后給它添加一些值：

var subject = new Rx.Subject();

subject.subscribe({
    next: v => console.log("observerA: " + v)
});

subject.subscribe({
    next: v => console.log("observerB: " + v)
});

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

執(zhí)行后的輸出：

observerA: 1
observerB: 1
observerA: 2
observerB: 2

由于 Subject 也是一個(gè) Observer ，所以你可以把他傳遞給一個(gè) Observable 的subscribe方法：

var subject = new Rx.Subject();

subject.subscribe({
    next: v => console.log("observerA: " + v)
});

subject.subscribe({
    next: v => console.log("observerB: " + v)
});

var observable = Rx.Observable.from([1,2,3]);

observable.subscribe(subject); // 通過 Subject 來訂閱這條 Observable

執(zhí)行后輸出：

observerA: 1
observerB: 1
observerA: 2
observerB: 2
observerA: 3
observerB: 3

通過上面的方法，我們基本上就借助 Subject 把一個(gè)單播的 Observable 轉(zhuǎn)換成了多播的。這個(gè)示例僅僅是演示了一種將一個(gè)執(zhí)行上下文分享給多個(gè)觀察者的方法。

在 Subject 類型下，還有一些比較特殊的 Subject 類型：BehaviorSubject，ReplaySubject，AsyncSubject。

一個(gè)‘多播’的 Observable 可以借助于 Subject 實(shí)現(xiàn)將數(shù)據(jù)通知給多個(gè)觀察者，然而單播的 Observable 僅僅只能把通知發(fā)送給一個(gè)觀察者。

_多播的 Observable 會(huì)使用 Subject 作為鉤子來實(shí)現(xiàn)將同一個(gè) Observable 的執(zhí)行傳遞給多個(gè)觀察者。_

這就是那些多播的操作符實(shí)現(xiàn)時(shí)的真實(shí)情況：觀察者訂閱底層的一個(gè) Subject，使用這個(gè) Subject 來訂閱產(chǎn)生源數(shù)據(jù)的 Observable。

下面這個(gè)例子和之前那個(gè)使用 subject 來訂閱 Observable 的例子非常類似：

var source = Rx.Observable.from([1,2,3]);
var subject = new Rx.Subject();
var multicasted = source.multicast(subject);

// 使用 subject.subscribe({...}) 實(shí)現(xiàn)
multicasted.subscribe({

   next: v => console.log("observerA: " + v)

});
multicasted.subscribe({

   next: v => console.log("observerB: " + v)

});

//使用 source.subscribe(subject) 實(shí)現(xiàn)

multicasted.connect();

上面代碼中的multicast方法返回的 Observable 看起來像一個(gè)普通的 Observable，但是當(dāng)它被訂閱時(shí)，執(zhí)行的方式卻和 Subject 一樣是多播的，同時(shí)這個(gè) Observable 還有一個(gè)connect方法，可以將多個(gè)流串聯(lián)在一起。

何時(shí)調(diào)用這個(gè)connect方法非常重要，因?yàn)樗鼪Q定了這條可以共享的流什么時(shí)間開始執(zhí)行。由于connect方法在內(nèi)部執(zhí)行了 source.subscribe(subject) ，因此它會(huì)返回一個(gè) Subscription ，可以通過它來取消這個(gè)共享的 Observable 的執(zhí)行。

通過手動(dòng)調(diào)用connect方法來得到 Subscription 多少顯得有些笨重，通常情況下，我們希望當(dāng)?shù)谝粋€(gè)觀察者抵達(dá)時(shí)可以自動(dòng)的調(diào)用connect方法，當(dāng)最后一個(gè)觀察者取消訂閱時(shí)自動(dòng)的取消這個(gè)共享 Observable 的執(zhí)行。

請(qǐng)考慮實(shí)現(xiàn)如以下列表所概述的訂閱：

第一個(gè)觀察者訂閱多播流。

多播流被connect。

0 被傳遞給第一個(gè)訂閱者。

第二個(gè)觀察者訂閱這條多播流。

1 被傳遞給第一個(gè)訂閱者。

1 被傳遞給第二個(gè)訂閱者。

第一個(gè)訂閱者取消訂閱多播流。

2 被 傳遞給第二個(gè)訂閱者。

第二個(gè)訂閱者取消訂閱多播流。

取消這條多播流的訂閱。

為了達(dá)到這個(gè)效果，我們需要顯式的調(diào)用connect方法：

var source = Rx.Observable.interval(500);
var subject = new Rx.Subject();
var multicasted = source.multicast(subject);
var subscription1, subscription2, subscriptionConnect;

subscription1 = multicasted.subscribe({
    next: v => console.log("observerA: " + v)
});

// 在這里我們需要顯示的調(diào)用connect方法以使第一個(gè)訂閱者可以開始接收值
subscriptionConnect = multicasted.connect();

setTimeout(() => {
    subscription2 = multicasted.subscribe({
        next: v => console.log("observerB: " + v)
    });
}, 600);

setTimeout(() => {
    subscription1.unsubscribe();
},1200);

// 在這里我們需要把多播流的訂閱取消掉，因?yàn)閺拇艘院笤僖矝]有訂閱者訂閱它了。
setTimeout(() => {
    subscription2.unsubscribe();
    subscriptionConnect.unsubscribe();
},2000);

我們可以通過使用 ConnectableObservable 的refCount方法來避免顯式的調(diào)用connect方法，這個(gè)方法會(huì)返回一個(gè) Observable 用來追蹤當(dāng)前有多少訂閱者正在訂閱多播流。當(dāng)訂閱者的數(shù)量從0增加到1時(shí)，它幫助我們調(diào)用connect方法以開始訂閱，當(dāng)訂閱者的數(shù)量從1變?yōu)?時(shí)，它幫助我們?nèi)∠嗛喴酝Ｖ苟嗖チ鞯睦^續(xù)執(zhí)行。

_refCount使得多播流在第一個(gè)訂閱者到達(dá)時(shí)開始執(zhí)行，在最后一個(gè)訂閱者離開后停止執(zhí)行。_

請(qǐng)看下面的例子：

var source = Rx.Observable.interval(500);
var subject = new Rx.Subject();
var refCount = source.multicast(subject).refCount();
var subscription1, subscription2, subscriptionConnect;

// 這次調(diào)用會(huì)執(zhí)行connect方法
console.log("observerA subscribed");
subscription1 = refCounted.subscribe({
    next: v => console.log("observerA: " + v)
});

setTimeout(() => {
    console.log("observerB subscribed");
    subscription2 = refCounted.subscribe({
        next: v => console.log("observerB: " + v)
    })
},600);

setTimeout(() => {
    console.log("observerA unsubscribed");
    subscription1.unsubscribe();
}, 1200);

// 這里會(huì)調(diào)用多播流的unsubscribe方法
setTimeout(() => {
    console.log("observerB unsubscribed");
    subscription2.unsubscribe();
},2000);

執(zhí)行后的輸出：

observerA subscribed
observerA: 0
observerB subscribed
observerA: 1
observerB: 1
observerA unsubscribed
observerB: 2
observerB unsubscribed

refCount方法僅存在于 ConnectableObservable 上，它返回的是一個(gè)新的 Observable 而不是另一個(gè) ConnectableObservable 。

BehaviorSubject 是 Subject 的一個(gè)變種上，關(guān)于它的一個(gè)重要概念是’當(dāng)前值‘。它會(huì)把最后一次發(fā)送給訂閱者的值保存下來，當(dāng)另外一個(gè)訂閱者開始訂閱時(shí)，它會(huì)把這個(gè)值立即發(fā)送給新的訂閱者。

_BehaviorSubject 對(duì)于表示隨時(shí)間變化的值是非常有用的。例如，表示生日的事件流可以用 Subject，但是一個(gè)人的年齡的事件流可以用 BehaviorSubject。_

下面的例子中，BehaviorSubject初始化了一個(gè)值0，當(dāng)?shù)谝粋€(gè)訂閱者開始訂閱時(shí)，這個(gè)值被發(fā)送出去。接下來當(dāng)?shù)诙€(gè)訂閱者開始訂閱時(shí)，將會(huì)接收到值2，即使這個(gè)值已經(jīng)被發(fā)送過了。

var subject = new Rx.BehaviorSubject(0);

subject.subscribe({
    next: v => console.log("observerA: " + v)
});

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

subject.subscribe({
    next: v => console.log("observerB: " + v)
});

subject.next(3);

執(zhí)行后的輸出：

observerA: 0
observerA: 1
observerA: 2
observerB: 2
observerA: 3
observerB: 3

ReplaySubject 和 BehaviorSubject 很類似，它們都可以把過去的值發(fā)送給新的訂閱者，不同的是它可以把 Observable 執(zhí)行時(shí)產(chǎn)生的一部分值記錄下來。

_ReplaySubject 記錄 Observable 執(zhí)行時(shí)的多個(gè)值，并將它們傳遞給新加入的訂閱者。_

在創(chuàng)建 ReplaySubject 時(shí)，我們可以指定需要重復(fù)發(fā)送的值的數(shù)量：

var subject = new Rx.ReplaySubject(3); // 傳遞3個(gè)值給新加入的訂閱者

subject.subscribe({
    next: (v) => console.log("observerA: " + v)
});

subject.next(1);
subject.next(2);
subject.next(3);
subject.next(4);

subject.subscribe({
    next: (v) => console.log("observerB: " + v)
});

subject.next(5);

執(zhí)行后的輸出：

observerA: 1
observerA: 2
observerA: 3
observerA: 4
observerB: 2
observerB: 3
observerB: 4
observerA: 5
observerB: 5

還可以指定一個(gè)以毫秒為單位的時(shí)間，配合記錄的數(shù)量共同決定究竟有個(gè)多少個(gè)值需要被重復(fù)發(fā)送。

下面的這個(gè)例子，使用了一個(gè)比較大的數(shù)量，但是時(shí)間上只限制了只發(fā)送500毫秒內(nèi)的值：

var subject = new ReplaySubject(100, 500/* windowTime */);

subject.subscribe({
    next: v => console.log("observerA: " + v)
});

var i = 1;
setInterval(() => subject.next(i++), 200);

setTimeout(() => {
    subject.subscribe({
        next: v => console.log("observerB: " + v)
    });
},1000);

第二個(gè)訂閱者只能接收到最近的500毫秒內(nèi)發(fā)出的值，下面是執(zhí)行后的輸出，

observerA: 1
observerA: 2
observerA: 3
observerA: 4
observerA: 5
observerB: 3
observerB: 4
observerB: 5
observerA: 6
observerB: 6
...

它也是 Subject 的一個(gè)變種，AsyncSubject僅在流執(zhí)行結(jié)束后把最后一個(gè)值發(fā)送給它的訂閱者。

var subject = new Rx.AsyncSubject();

subject.subscribe({
    next: v => console.log("observerA: " + v)
});

subject.next(1);
subject.next(2);
subject.next(3);
subject.next(4);

subject.subscribe({
    next: v => console.log("observerB: " + v);
});

subject.next(5);
subject.complete();

執(zhí)行后的輸出：

observerA: 5
observerB: 5

AsyncSubject 的行為與last操作符的行為非常相似，都是在等待完成后再發(fā)送一個(gè)值。你應(yīng)該還記得之前提到的當(dāng) Observable 發(fā)出Complete通知或Error通知后就不能再發(fā)送值，AsyncSubject看起來違背了這個(gè)原則，其實(shí)不然，我們可以看一下它的源碼：

constructor() {
    super(...arguments);
    this.value = null;
    this.hasNext = false;
    this.hasCompleted = false;
}
complete() {
    this.hasCompleted = true;
    if (this.hasNext) {
        super.next(this.value);
    }
    super.complete();
}

這里只摘錄出了AsyncSubject的構(gòu)造函數(shù)和它的complete方法，首先AsyncSubject是繼承自Subject的，所以這里的super類就是Subject，那么就很明顯了，在AsyncSubject實(shí)例上調(diào)用complete方法時(shí)并沒有違背之前提到的原則，依然是先發(fā)出了Next通知，最后才發(fā)出Complete通知。

雖然 Observable 是構(gòu)建rxjs的基石，但是真正精彩的部分應(yīng)該是操作符。操作符的組合使用使得以聲明式的方式解決復(fù)雜的異步場(chǎng)景變得非常簡(jiǎn)單，它構(gòu)成rxjs必不可少的一部分。

操作符是 Observable 類上的方法，例如： .map(...)，.filter(...)，.merge(...)等。當(dāng)我們調(diào)用它們時(shí)，并不會(huì)去改變已有 Observable 實(shí)例，取而代之的是創(chuàng)建一個(gè)全新的 Observable ，它是訂閱邏輯是基于第一個(gè) Observable 的。

_操作符是一個(gè)純函數(shù)，它不會(huì)改變已有的 Observable 實(shí)例，而是基于當(dāng)前的Observable 創(chuàng)建一個(gè)新的 Observable。_

理解操作符時(shí)非常重要的一點(diǎn)是時(shí)刻謹(jǐn)記它是一個(gè)純函數(shù)，它接受一個(gè) Observable 作為輸入然后生成一個(gè)新的 Observable 作為輸出。當(dāng)訂閱輸出 Observable 的同時(shí)也就訂閱了輸入 Observable 。

下面的例子中，我們自定義了一個(gè)操作函數(shù)，它將輸入的每一個(gè)值都乘以10：

function multiplyByTen(input) {
    var output = Rx.Observable.create(function subscribe(observer) {
        input.subscribe({
            next: v => observer.next(10 * v),
            error: err => observer.error(error),
            complete: () => observer.complete()
        });
    });

    return output;
}

var input = Rx.Observable.from([1,2,3,4]);
var output = multiplyByTen(input);
output.subscribe(x => console.log(x));

執(zhí)行后的輸出：

10
20
30
40

注意：我們只是訂閱了輸出-——output，但這同時(shí)導(dǎo)致input被訂閱，這稱之為’操作符的鏈?zhǔn)接嗛啞?/p> 實(shí)例的操作符 VS 靜態(tài)操作符

通常在提到操作符時(shí)，我們指的是 Observable 實(shí)例上的操作符。假如上面例子中我們定義的操作函數(shù) multiplyByTen 是一個(gè)操作符的話，它看起來應(yīng)該是這樣：

Rx.Observable.prototype.multiplyByTen = function multiplyByTen() {
    var input = this;
    return Rx.Observable.create(function subscribe(observer) {
        input.subscribe({
            next: v => observer.next(10 * v),
            error: err => observer.error(err),
            complete: () => observer.complete()
        });
    });
}

_實(shí)例操作符是使用this關(guān)鍵字來推斷輸入 Observable 的函數(shù)。_

在這里輸入的 Observable 不再通過 multiplyByTen 的參數(shù)獲得，而是通過this關(guān)鍵字來獲取，這就允許我們可以像下面這樣鏈?zhǔn)降恼{(diào)用：

var observable = Rx.Observable.from([1,2,3,4]).multiplyByTen();

observable.subscribe(x = console.log(x));

除了實(shí)例操作符，還有靜態(tài)操作符，它們是直接定義在 Observable 類上的靜態(tài)方法。靜態(tài)操作符不會(huì)使用this關(guān)鍵字來推斷輸入，它的輸入完全依賴于輸入的參數(shù)。

_靜態(tài)操作符時(shí)附加在 Observable 類上的靜態(tài)函數(shù)，通常用來從頭創(chuàng)建 Observable。_

常見的靜態(tài)操作符都是一些可創(chuàng)建類型的操作符，與通常情況下操作符接收輸入流，輸出輸出流不同，它們大多接收一些非 Observable 類型的參數(shù)，例如一個(gè)數(shù)字，然后創(chuàng)建出一條流。

var observable = Rx.Observable.interval(1000 /* 毫秒數(shù) */);

另外的例子就是create，我們?cè)谥暗睦又幸呀?jīng)多次使用過了。

然后，靜態(tài)操作符并非只是那些簡(jiǎn)單創(chuàng)建流的操作符，一些組合類的操作符也可以有它的靜態(tài)類型，例如：merge，combineLatest，concat等。這樣做的意義在于我們接收多個(gè)流作為輸入而不僅僅是一個(gè)。

var observable1 = Rx.Observable.interval(1000);
var observable2 = Rx.Observable.interval(400);

var merge = Rx.Observable.merge(observable1, observable2);

單純靠文字可能還不足解釋的清楚操作符是如何工作的，許多操作符是與時(shí)間相關(guān)的，它們可能以不同的方式來影響值的發(fā)射，比如：delay，throttle，sample等。對(duì)于這些操作符來說圖表的形式會(huì)更加直觀。彈珠圖可以模擬出操作符是如何工作的，可以直觀的表現(xiàn)出輸入流，操作符，參數(shù)與輸出流之間的聯(lián)系。

下面來詳細(xì)解釋彈珠圖各部分的含義：

// 這條從左到右的橫線代表隨時(shí)間的推移，輸入流的執(zhí)行過程。
// 橫線上的值代表從流上發(fā)射出的值
// 橫線尾部的豎線代表complete通知執(zhí)行的時(shí)間點(diǎn)，表示這條流已經(jīng)成功的執(zhí)行完成。
----------4------6--------------a-------8-------------|---->

            multipleByTen // 使用的操作符

// 這條從左到右的橫線代表經(jīng)過操作符轉(zhuǎn)換后的輸出流。
// 橫線尾部的X代表在這個(gè)時(shí)間點(diǎn)上流發(fā)生了錯(cuò)誤，至此之后不應(yīng)該再有 Next 通知或 Complete 通知從流上發(fā)出。
---------40-----60--------------X--------------------------->

在整個(gè)文檔中，我們都使用這種彈珠圖來解釋操作符是如何工作的。這種彈珠圖在其它場(chǎng)景中也是非常有用的，比如在白板演示或者單元測(cè)試中。

如果你明確自己需要解決的問題，但是不清楚應(yīng)該使用哪一個(gè)操作符，可以在官網(wǎng)的 Manual > Operators > Choose an operator 中通過選擇符合問題的描述來找到你所需要的操作符。

為了解決不同的問題，rxjs針對(duì)性的設(shè)計(jì)了各種各樣的操作符，大體上可以分為幾類：創(chuàng)建型、轉(zhuǎn)換型、過濾型、組合型、多播型、錯(cuò)誤處理型、工具類等等。

關(guān)于操作符，將會(huì)在專門的文檔中進(jìn)行解釋。

Scheduler決定了subscription什么時(shí)候開始以及什么時(shí)候開始分發(fā)值。它由3個(gè)部分組成：

一個(gè)Scheduler就是一段數(shù)據(jù)結(jié)構(gòu)：它知道如何去存儲(chǔ)數(shù)據(jù)，以及基于優(yōu)先級(jí)或其它標(biāo)準(zhǔn)來排列任務(wù)的順序。

一個(gè)Scheduler就是一個(gè)執(zhí)行環(huán)境：決定什么時(shí)間在什么樣的環(huán)境下去執(zhí)行任務(wù)， 是立即執(zhí)行還是在回調(diào)中執(zhí)行，亦或是在定時(shí)器到達(dá)時(shí)，還是下一次事件循環(huán)開始時(shí)執(zhí)行。

Scheduler有自己的虛擬時(shí)鐘：被調(diào)度任務(wù)的執(zhí)行時(shí)間只由虛擬時(shí)鐘決定。只會(huì)它的虛擬時(shí)鐘的環(huán)境中執(zhí)行。

null： 按正常的規(guī)則發(fā)送值。

queue: 按當(dāng)前事件隊(duì)列中的順序來發(fā)送值。

asap：按更小的單元上的事件隊(duì)列中的順序發(fā)送值。

async: 按異步順序來發(fā)送值。

可以傳入scheduler的操作符：

bindCallback

bindNodeCallback

combineLatest

concat

empty

from

fromPromise

interval

merge

of

range

throw

timer