摘要：函數(shù)調(diào)用后同步計(jì)算并返回單一值生成器函數(shù)遍歷器遍歷過程中同步計(jì)算并返回個(gè)到無窮多個(gè)值異步執(zhí)行中返回或者不返回單一值同步或者異步計(jì)算并返回個(gè)到無窮多個(gè)值是函數(shù)概念的拓展既不像，也不像是。如果不調(diào)用函數(shù)，就不會(huì)執(zhí)行如果如果不訂閱，也不會(huì)執(zhí)行。

Observable（可觀察對象）是基于推送（Push）運(yùn)行時(shí)執(zhí)行（lazy）的多值集合。下方表格對Observable進(jìn)行了定位（為解決基于推送的多值問題）：

例：當(dāng)observable被訂閱后，會(huì)立即（同步地）推送1， 2， 3 三個(gè)值；1秒之后，繼續(xù)推送4這個(gè)值，最后結(jié)束（推送結(jié)束通知）：

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

為得到observable推送的值，我們需要訂閱（subscribe）這個(gè)Observable：

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í)行后，將在控制臺輸出如下結(jié)果：

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

拉取和推送是數(shù)據(jù)生產(chǎn)者和數(shù)據(jù)消費(fèi)者之間通信的兩種不同機(jī)制。

何為拉取？ 在拉取系統(tǒng)中，總是由消費(fèi)者決定何時(shí)從生產(chǎn)者那里獲得數(shù)據(jù)。生產(chǎn)者對數(shù)據(jù)傳遞給消費(fèi)者的時(shí)間毫無感知（被動(dòng)的生產(chǎn)者，主動(dòng)的消費(fèi)者）。

JavaScript函數(shù)是典型的拉取系統(tǒng)：函數(shù)是數(shù)據(jù)的生產(chǎn)者，對函數(shù)進(jìn)行調(diào)用的代碼（消費(fèi)者）從函數(shù)調(diào)用后的返回值中拉取單值進(jìn)行消費(fèi)。

// 函數(shù)是數(shù)據(jù)的生產(chǎn)者
let getLuckyNumber = function() {
    return 7;
};

/* let代碼段是數(shù)據(jù)的消費(fèi)者，
 * getLuckyNumber對調(diào)用時(shí)間毫無感知。 
 */
let luckNumber = getLuckyNumber();

ES2015 引入了的 生成器函數(shù) | 遍歷器 (function*)同樣是基于拉取的系統(tǒng)： 調(diào)用 iterator.next()的代碼段是消費(fèi)者，它可以從生成器函數(shù)中拉取多個(gè)值。

function* getLessThanTen() {
  var i = 0;
  while(i < 11) {
    yield i++;
  }
}

// 生產(chǎn)者
let iterator = getLessThanTen();

// 消費(fèi)者
iterator.next(); // Object {value: 0, done: false}
iterator.next(); // Object {value: 1, done: false}

何為推送？ 在推送系統(tǒng)中生產(chǎn)者決定何時(shí)向消費(fèi)者傳遞數(shù)據(jù)，消費(fèi)者對何時(shí)收到數(shù)據(jù)毫無感知（被動(dòng)的消費(fèi)者）。

現(xiàn)代JavaScript中Promise是典型的推送系統(tǒng)。作為數(shù)據(jù)生產(chǎn)者的Promise通過resolve()向數(shù)據(jù)消費(fèi)者——回調(diào)函數(shù)傳遞數(shù)據(jù)：與函數(shù)不同，Promise決定向回調(diào)函數(shù)推送值的時(shí)間。

RxJS在 JavaScript 中引入了Observable(可觀察對象)這個(gè)新的推送系統(tǒng)。Observable是多數(shù)據(jù)值的生產(chǎn)者，向Observer(被動(dòng)的消費(fèi)者)推送數(shù)據(jù)。

函數(shù) 調(diào)用后同步計(jì)算并返回單一值

生成器函數(shù) | 遍歷器 遍歷過程中同步計(jì)算并返回0個(gè)到無窮多個(gè)值

Promise 異步執(zhí)行中返回或者不返回單一值

Observable 同步或者異步計(jì)算并返回0個(gè)到無窮多個(gè)值

Observable既不像EventEmitter，也不像是Promise。Observable 中的 Subject 進(jìn)行多路推送時(shí)與 EventEmitter 行為上有些類似，但是實(shí)際上Observable與EventEmitter并不相同。

Observable 更像是一個(gè)不需要傳入?yún)?shù)的函數(shù)，它拓展了函數(shù)的概念使得它可以返回多個(gè)值。

看看下面的例子：

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

var x = foo.call(); // same as foo()
console.log(x);
var y = foo.call(); // same as foo()
console.log(y);

輸出結(jié)果如下:

"Hello"
42
"Hello"
42

通過Observable可以實(shí)現(xiàn)同樣的行為：

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

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

輸出結(jié)果相同：

"Hello"
42
"Hello"
42

不論Observable還是函數(shù)都是在運(yùn)行時(shí)進(jìn)行求值計(jì)算的。如果不調(diào)用函數(shù)，console.log("Hello")就不會(huì)執(zhí)行；如果如果不subscribe（訂閱）Observable，console.log("Hello")也不會(huì)執(zhí)行。此外，調(diào)用或者訂閱都是獨(dú)立的：兩次調(diào)用產(chǎn)生兩個(gè)獨(dú)立的作用域，兩次訂閱同樣會(huì)產(chǎn)生兩個(gè)獨(dú)立的作用域。EventEmitter總是在同一個(gè)作用域中，發(fā)射前也不會(huì)在意自己是否已經(jīng)被訂閱；Observable不會(huì)被共享而產(chǎn)生副作用，并且總是在被訂閱時(shí)才執(zhí)行。

訂閱Observable與調(diào)用函數(shù)類似。

一些人認(rèn)為Observable總是是異步的，這個(gè)觀點(diǎn)并不正確，如果在控制臺log函數(shù)中調(diào)用函數(shù)：

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

顯然可以看到以下輸出：

"before"
"Hello"
42
"after"

Observable的行為完全一樣：

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

輸出結(jié)果為：

"before"
"Hello"
42
"after"

訂閱 foo完全是同步的，與函數(shù)的調(diào)用一樣。

Observable可以異步或者同步地產(chǎn)生數(shù)據(jù)。

那Observable 與函數(shù)的不同之處在哪里？ Observable可以在一個(gè)時(shí)間過程中‘返回’多個(gè)值，而函數(shù)卻不能。在函數(shù)中你不可以這么做：

function foo() {
  console.log("Hello");
  return 42;
  return 100; // 這個(gè)語句永遠(yuǎn)不會(huì)被執(zhí)行。
}

雖然函數(shù)只能有一個(gè)返回值，但是在Observable中你完全可以這么做：

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

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

輸出結(jié)果如下:

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

你甚至可以異步地返回值：

var foo = Rx.Observable.create(function (observer) {
  console.log("Hello");
  observer.next(42);
  observer.next(100);
  observer.next(200);
  setTimeout(() => {
    observer.next(300); // happens asynchronously
  }, 1000);
});

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

輸出結(jié)果：

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

結(jié)論：

func.call() 意味著“同步地給我一個(gè)值”

observable.subscribe() 意味著“不管是同步或者異步，給我一些值”

通過使用 Rx.Observable.create 或者是創(chuàng)建操作符，創(chuàng)建一個(gè)Observable； Observable 被 Observer（觀察者） 訂閱； 在執(zhí)行時(shí) 向觀察者發(fā)送next / error / complete 通知；同時(shí)執(zhí)行過程可以被 終止。
Observable 類型的實(shí)例具備了以上四個(gè)方面的特性，與其他類型如：Observer 和 Subscription 緊密相關(guān)。

我們重點(diǎn)關(guān)注以下四個(gè)方面：

創(chuàng)建

訂閱

執(zhí)行

終止

Rx.Observable.create 是 Observable 構(gòu)造函數(shù)的別名，接受一個(gè)參數(shù)： subscribe函數(shù)。

以下例子會(huì)創(chuàng)建一個(gè)Observable，每一秒鐘向其訂閱者發(fā)射一個(gè)"hi" 字符串。

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

除了使用create創(chuàng)建Observable，我們通常還使用創(chuàng)建操作符, 如 of，from， interval, 等來創(chuàng)建Observable。

上面例子中，subscribe函數(shù)是定義Observable最重要的部分。我們接下來了解訂閱的含義。

上面例子中的observable 可以以如下方式 訂閱 ：

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

observable.subscribe 和 Observable.create(function subscribe(observer) {...})中的subscribe 同名并非巧合。雖然在Rx中它們不是同一個(gè)對象，但是在工程中，我們可以在概念上視兩者為等價(jià)物。

調(diào)用subscribe的觀察者并不會(huì)共享同一個(gè)Observable。觀察者調(diào)用observable.subscribe 時(shí)，Observable.create(function subscribe(observer) {...})中的subscribe會(huì)在調(diào)用它的觀察者作用域中執(zhí)行。每一次observable.subscribe的調(diào)用，都是彼此獨(dú)立的。

訂閱Observable如同調(diào)用函數(shù)，需要提供相應(yīng)的回調(diào)方法。

訂閱機(jī)制與處理事件的addEventListener / removeEventListenerAPI完全不同。通過observable.subscribe，觀察者并不需要在Observable中進(jìn)行注冊，Observable也不需要維護(hù)訂閱者的列表。

訂閱后便進(jìn)入了Observable的執(zhí)行階段，在執(zhí)行階段值和事件將會(huì)被傳遞給觀察者供其消費(fèi)。

只有在被訂閱之后Observable才會(huì)執(zhí)行，執(zhí)行的邏輯在Observable.create(function subscribe(observer) {...})中描述，執(zhí)行后將會(huì)在特定時(shí)間段內(nèi)，同步或者異步地成產(chǎn)多個(gè)數(shù)據(jù)值。

Observable在執(zhí)行過程中，可以推送三種類型的值：

"Next" 通知： 實(shí)際產(chǎn)生的數(shù)據(jù)，包括數(shù)字、字符串、對象等

"Error" 通知：一個(gè)JavaScript錯(cuò)誤或者異常

"Complete" 通知：一個(gè)不帶有值的事件

“Next” 通知是最重要和常用的類型：表示事件傳遞給觀察者的數(shù)據(jù)。錯(cuò)誤和完成通知僅會(huì)在執(zhí)行階段推送其一，并不會(huì)同時(shí)推送錯(cuò)誤和完成通知。

通過所謂的“Observable語法”或者“契約”可以最好地表達(dá)這個(gè)規(guī)則，“Observable語法”借助于正則表達(dá)式：

next*(error|complete)?

在Observable的執(zhí)行過程中，0個(gè)或者多個(gè)“Next”通知會(huì)被推送。在錯(cuò)誤或者完成通知被推送后，Observable不會(huì)再推送任何其他通知。

下面代碼展示了Observable 在執(zhí)行過程中推送3個(gè)“Next” 通知然后結(jié)束：

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

Observable 嚴(yán)格遵守 Observable 契約，后面值為4的“Next” 通知永遠(yuǎn)不會(huì)被推送：

var observable = Rx.Observable.create(function subscribe(observer) {
  observer.next(1);
  observer.next(2);
  observer.next(3);
  observer.complete();
  observer.next(4); // 由于違法契約，4不會(huì)被推送
});

使用try/catch塊包裹 subscribe 代碼是一個(gè)很贊的想法，如果捕獲了異常，可以推送錯(cuò)誤通知：

var observable = Rx.Observable.create(function subscribe(observer) {
  try {
    observer.next(1);
    observer.next(2);
    observer.next(3);
    observer.complete();
  } catch (err) {
    observer.error(err); // 捕獲異常后推送錯(cuò)誤通知
  }
});

Observable的執(zhí)行可能是無限的，作為觀察者需要主動(dòng)中斷執(zhí)行：我們需要特定的API去終止執(zhí)行過程。因?yàn)樘囟ǖ挠^察者都有特定的執(zhí)行過程，一旦觀察者獲得想要的數(shù)據(jù)后就需要終止執(zhí)行過程以免帶來計(jì)算時(shí)對內(nèi)存資源的浪費(fèi)。

在observable.subscribe被調(diào)用時(shí)，觀察者會(huì)與其執(zhí)行作用域綁定，同時(shí)返回一個(gè)Subscription類型的對象：

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

Subscription對象表示執(zhí)行過程，通過極簡的API，你可以終止執(zhí)行過程。詳情請閱讀Subscription 相關(guān)文檔。通過調(diào)用subscription.unsubscribe() 你可以終止執(zhí)行過程：

var observable = Rx.Observable.from([10, 20, 30]);
var subscription = observable.subscribe(x => console.log(x));
// Later:
subscription.unsubscribe();

在Observable被訂閱后，代表執(zhí)行過程的Subscription 對象將被返回。對其調(diào)用unsubscribe()就可以終止執(zhí)行。

每一個(gè)Observable都需要在 create()的創(chuàng)建過程中定義終止的邏輯。在function subscribe()中返回自定義的unsubscribe就可以實(shí)現(xiàn)。

下面的例子說明了如何在終止后釋放setInterval的句柄：

var observable = Rx.Observable.create(function subscribe(observer) {
  // 獲得定時(shí)函數(shù)的句柄
  var intervalID = setInterval(() => {
    observer.next("hi");
  }, 1000);
  
  // 提供終止方法釋放定時(shí)函數(shù)的句柄
  return function unsubscribe() {
    clearInterval(intervalID);
  };
});

類似于observable.subscribe 和 Observable.create(function subscribe() {...})的關(guān)系，我們在subscribe中返回的 unsubscribe 也與subscription.unsubscribe在概念上等價(jià)。事實(shí)上，如果我們除去Rx的包裝，純粹的JavaScript代碼簡單清晰：

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

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

// 一段時(shí)間后:
unsubscribe(); // 終止

使用Observable、 Observer 和 Subscription這些概念的原因是，我們可以在Observable 契約之下安全、兼容地調(diào)用操作符。