摘要：通過(guò)執(zhí)行和可以向訂閱者推送不同的通知。之后，執(zhí)行過(guò)程可能被處理掉。當(dāng)調(diào)用并得到觀察者時(shí)，在中傳入的函數(shù)將會(huì)被執(zhí)行。每次執(zhí)行都會(huì)觸發(fā)一個(gè)多帶帶針對(duì)當(dāng)前的運(yùn)行邏輯。通知不發(fā)出任何值，表示流的結(jié)束。

原文:http://reactivex.io/rxjs/manu...

Rx.Observalbe.create()或者創(chuàng)建操作符，可以 創(chuàng)建（created） Observable流。
Observer則可以 訂閱（subscribed） 這個(gè)流。
通過(guò) 執(zhí)行（execute） next()、error()和complete()可以向訂閱者推送不同的通知。
之后，執(zhí)行過(guò)程可能被 處理掉（disposed） 。
這四個(gè)方面都被集成在Observable實(shí)例當(dāng)中，但是也有一些方面與其他類(lèi)型有關(guān)，比如Observer和Subscription。

Observable的核心關(guān)注點(diǎn)是:

創(chuàng)建Observable流

訂閱Observable流

執(zhí)行Observable流

終止Observable流

Rx.Observable.create可以說(shuō)是Observable構(gòu)造函數(shù)的別名，他可以接受一個(gè)參數(shù):subscribe函數(shù)。

以下的例子創(chuàng)建了一個(gè)Observable流，每秒鐘向Observer發(fā)出一個(gè)字符串類(lèi)性值hi。

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

Observables流可以使用create()創(chuàng)建，但是通常我們會(huì)使用所謂的創(chuàng)建操作符，像of(),from(),interval()等等。

在上面的例子中，訂閱函數(shù)(subscribe function)是描述Observalbe最重要的部分。那么，讓我來(lái)看看何謂訂閱。

在例子中，Observalbe的實(shí)例observable可以被訂閱，像這樣:

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

也許你會(huì)注意到，observable.subscribe()和subscribe函數(shù)在Rx.Observable.create(function subscribe(observer){...})中使用了相同的名字，這并不是巧合。
在庫(kù)中，他們是不同的，但在實(shí)際使用中，你可以認(rèn)為他們?cè)诟拍钌鲜窍嗟鹊摹?/p>

Observable不在多個(gè)Observer之間共享subscribe。當(dāng)調(diào)用observable.subscribe()并得到觀察者時(shí)，在Rx.Observable.create(function subscribe(observer){...})中傳入的函數(shù)將會(huì)被執(zhí)行。每次執(zhí)行observable.subscribe()都會(huì)觸發(fā)一個(gè)多帶帶針對(duì)當(dāng)前Observer的運(yùn)行邏輯。

訂閱一個(gè)Observable流就像調(diào)用一個(gè)函數(shù)，流中的數(shù)據(jù)將會(huì)被傳遞給回調(diào)函數(shù)中。

一個(gè)subscribe函數(shù)被調(diào)用將會(huì)開(kāi)啟一個(gè)Observable執(zhí)行流(Observable execution)，向觀察者們輸出流中的值或者事件。

代碼Rx.Observable.create(function subscribe(observer){...})代表了一個(gè)“Observable流”，由于惰性計(jì)算，只用當(dāng)有Observer訂閱流時(shí)，函數(shù)才會(huì)被執(zhí)行。
執(zhí)行過(guò)程中隨著時(shí)間線產(chǎn)生多個(gè)數(shù)據(jù)，方式是同步或異步二選一。

有三個(gè)類(lèi)型的值會(huì)在執(zhí)行流中發(fā)出:

"Next" 通知：發(fā)出一個(gè)值，比如數(shù)字，字符串，對(duì)象等等。

"Error"通知:發(fā)出一個(gè)js錯(cuò)誤或者異常。

Complete通知：不發(fā)出任何值，表示流的結(jié)束。

Next通知是最重要也是最常用的類(lèi)型：他代表了實(shí)際推送給Observer的值。Error和Complete通知只會(huì)在執(zhí)行流中發(fā)出一次，要么是Error，要么是Complete。

用正表達(dá)式的規(guī)則可以很好的表達(dá)這種所謂的Observable語(yǔ)法和約定：

next*(error|complete)?

在一個(gè)Observable執(zhí)行流中，會(huì)發(fā)出0到無(wú)限個(gè)Next通知。而一旦Error或者Complete通知被發(fā)出，執(zhí)行流將不會(huì)再推送任何消息。

下面的例子展示了一個(gè)推送了3個(gè)Next并Complete的流：

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

Observables會(huì)嚴(yán)格遵守Observable約定，所以下面的代碼將不會(huì)推送值4:

var observable = Rx.Observable.create(function subscribe(observer) {
  observer.next(1);
  observer.next(2);
  observer.next(3);
  observer.complete();
  observer.next(4); // Is not delivered because it would violate the contract
});

在訂閱函數(shù)中使用try/catch捕獲可能拋出的異常，也是一個(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); // delivers an error if it caught one
  }
});

Observable流的執(zhí)行時(shí)間線可能是無(wú)限長(zhǎng)的，但通常我們只用到有限的時(shí)間段和觀察者處理業(yè)務(wù)，因此，我們需要一種中斷流執(zhí)行的API。
由于一個(gè)執(zhí)行過(guò)程對(duì)于每個(gè)Observer是獨(dú)有的，一旦Observer接收到值，那么也必然需要一種中斷執(zhí)行的方式，從而可以節(jié)省計(jì)算性能和內(nèi)存空間。

當(dāng)observable.subscribe()被調(diào)用，Observer將被附加到新創(chuàng)建的Observable執(zhí)行過(guò)程中，同時(shí)返回了一個(gè)對(duì)象,Subscription：

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

Subscription代表了一個(gè)持續(xù)執(zhí)行的過(guò)程，并且有一套最小化的API允許你中斷流的執(zhí)行過(guò)程。可以從這里進(jìn)一步了解Subscription類(lèi)型。以下例子展示了使用subscription.unsubscribe()中斷持續(xù)執(zhí)行的過(guò)程：

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

當(dāng)你訂閱流就可以獲取一個(gè)Subscription，代表了持續(xù)執(zhí)行的過(guò)程。調(diào)用unsubscribe()就可以中斷執(zhí)行過(guò)程。

當(dāng)我們使用create()創(chuàng)建一個(gè)Observable流時(shí)，每一個(gè)Observable都必須定義它如何處理獲取到的資源的處理方式。你可以通過(guò)在subscribe()函數(shù)中返回一個(gè)自定義的unsubscribe函數(shù)，達(dá)到這個(gè)目的。

舉個(gè)例子，以下展示了如何中斷一個(gè)使用setInterval()執(zhí)行interval的過(guò)程:

var observable = Rx.Observable.create(function subscribe(observer) {
  // Keep track of the interval resource
  var intervalID = setInterval(() => {
    observer.next("hi");
  }, 1000);

  // Provide a way of canceling and disposing the interval resource
  return function unsubscribe() {
    clearInterval(intervalID);
  };
});

就像observable.subscribe()類(lèi)似Observable.create(function subscribe(){..})一樣，我們從subscribe()返回的unsubscribe()也概念性的等同于subscription.unsubscribe()。
事實(shí)上，如果我們移除與響應(yīng)式編程相關(guān)的概念，剩下的就是直白的js代碼了：

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

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

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

// Later:
unsubscribe(); // dispose the resources

我們使用Rx，包括Observable、Observer和Subscription，其原因就是為了使用這些安全(就如Observable約定的)和可組合的操作符。