摘要：通過執行和可以向訂閱者推送不同的通知。之后，執行過程可能被處理掉。當調用并得到觀察者時，在中傳入的函數將會被執行。每次執行都會觸發一個多帶帶針對當前的運行邏輯。通知不發出任何值，表示流的結束。

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

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

Observable的核心關注點是:

創建Observable流

訂閱Observable流

執行Observable流

終止Observable流

Rx.Observable.create可以說是Observable構造函數的別名，他可以接受一個參數:subscribe函數。

以下的例子創建了一個Observable流，每秒鐘向Observer發出一個字符串類性值hi。

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

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

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

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

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

也許你會注意到，observable.subscribe()和subscribe函數在Rx.Observable.create(function subscribe(observer){...})中使用了相同的名字，這并不是巧合。
在庫中，他們是不同的，但在實際使用中，你可以認為他們在概念上是相等的。

Observable不在多個Observer之間共享subscribe。當調用observable.subscribe()并得到觀察者時，在Rx.Observable.create(function subscribe(observer){...})中傳入的函數將會被執行。每次執行observable.subscribe()都會觸發一個多帶帶針對當前Observer的運行邏輯。

訂閱一個Observable流就像調用一個函數，流中的數據將會被傳遞給回調函數中。

一個subscribe函數被調用將會開啟一個Observable執行流(Observable execution)，向觀察者們輸出流中的值或者事件。

代碼Rx.Observable.create(function subscribe(observer){...})代表了一個“Observable流”，由于惰性計算，只用當有Observer訂閱流時，函數才會被執行。
執行過程中隨著時間線產生多個數據，方式是同步或異步二選一。

有三個類型的值會在執行流中發出:

"Next" 通知：發出一個值，比如數字，字符串，對象等等。

"Error"通知:發出一個js錯誤或者異常。

Complete通知：不發出任何值，表示流的結束。

Next通知是最重要也是最常用的類型：他代表了實際推送給Observer的值。Error和Complete通知只會在執行流中發出一次，要么是Error，要么是Complete。

用正表達式的規則可以很好的表達這種所謂的Observable語法和約定：

next*(error|complete)?

在一個Observable執行流中，會發出0到無限個Next通知。而一旦Error或者Complete通知被發出，執行流將不會再推送任何消息。

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

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

Observables會嚴格遵守Observable約定，所以下面的代碼將不會推送值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
});

在訂閱函數中使用try/catch捕獲可能拋出的異常，也是一個很不錯的做法：

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流的執行時間線可能是無限長的，但通常我們只用到有限的時間段和觀察者處理業務，因此，我們需要一種中斷流執行的API。
由于一個執行過程對于每個Observer是獨有的，一旦Observer接收到值，那么也必然需要一種中斷執行的方式，從而可以節省計算性能和內存空間。

當observable.subscribe()被調用，Observer將被附加到新創建的Observable執行過程中，同時返回了一個對象,Subscription：

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

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

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

當你訂閱流就可以獲取一個Subscription，代表了持續執行的過程。調用unsubscribe()就可以中斷執行過程。

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

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

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()類似Observable.create(function subscribe(){..})一樣，我們從subscribe()返回的unsubscribe()也概念性的等同于subscription.unsubscribe()。
事實上，如果我們移除與響應式編程相關的概念，剩下的就是直白的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約定的)和可組合的操作符。