• RxJS官方教程（一） 概览
• RxJS官方教程（二） Observable
• RxJS官方教程（三） Observable剖析
• RxJS官方教程（四） Observer和Subscription
• RxJS官方教程（五） Subject
• RxJS官方教程（六） 算子

Observable剖析

Observable可以使用Rx.Observable.create或 创建操作符 来创建，可以被 观察者Observer 订阅，可以执行next/ error/ complete方法通知观察者，并且它的执行也可以取消。这四个方面都在Observable中编码，但其中一些方面也与其他类型相关，如Observer和Subscription。

Observable核心关注点：

• 创建 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);
});

Observable可以通过create创建，但通常我们使用所谓的创建操作符来创建它。如of，from，interval等

在上面的例子中，subscribe函数是描述Observable最重要的部分。让我们来看看订阅的含义。

订阅Observable

可以订阅上例中的Observable observable，如下所示：

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

这不是一个巧合，observable.subscribe和Observable.create(function subscribe(observer) {...})中的名称subscribe是相同的，它们含义是不同的，但出于实际目的，您可以认为它们在概念上是平等的。

这说明对于同一个Observable，多个观察者Observer的observable.subscribe方法不是共享的。当Observer调用observable.subscribe时，Observable.create(function subscribe(observer) {...})中的函数将会执行。每一个Observer调用observable.subscribe都会触发自己的独立执行。

订阅Observable就像调用一个函数一样，提供一个传递数据的回调函数。

这与addEventListener/removeEventListener 等事件处理程序API完全不同。当调用observable.subscribe，给定的Observer未在Observable中注册为监听器。Observable甚至没有维护一份观察员名单。

subscribe方法可以很容易地启动一个“Observable execution”，并传递 值或事件 到订阅了该execution的Observer。

执行Observable

Observable.create(function subscribe(observer) {...})里面的代码表示一个“Observable execution”，这是一个只对每个观察者Observer发生的惰性计算。执行会随着时间的推移产生多个值，无论是同步还是异步。

Observable Execution可以提供三种类型的值：

• “Next”通知：发送一个值，如数字，字符串，对象等。
• “Error”通知：发送JavaScript错误或异常。
• “Complete”通知：不发送值。

Next通知是最重要和最常见的类型：它们表示传递给Observer的实际数据。在Observable Execution期间，Error和Complete通知只能发生一次，并且只能有其中一个。

这些约束最好用所谓的Observable Grammar或Contract表示，作为正则表达式写成：

next*(error|complete)?

在Observable Execution中，可以传递零到无限的Next通知。如果传递了Error或Complete通知，则之后不能传递任何其他内容。

以下是一个Observable执行的示例，它提供3个Next通知，然后Complete：

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

Observable严格遵守Observable Contract，因此以下代码不会提供Next通知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包裹subscribe内的代码，如果它捕获异常，它将传递错误通知：

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的执行

因为Observable 的执行可能是无限的，并且Observer想要在有限时间内中止执行是很常见的，所以我们需要一个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();

subscribe时，您将获得一个订阅Subscription，该Subscription表示正在进行的执行。可以调用unsubscribe()取消执行。

当使用create()创建Observable时，每个Observable都必须定义如何处置该执行的资源。您可以通过从function subscribe()内部返回自定义函数unsubscribe来实现。

例如，这是我们清除间隔执行setInterval的方式：

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类似，我们从subscribe返回的unsubscribe概念上类似于subscription.unsubscribe。事实上，如果我们抛开ReactiveX的这些概念，将留下相当简单的JavaScript。

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

我们使用像Observable，Observer和Subscription这样的Rx类型的原因是为了获得安全性（例如Observable Contract）和与操作符的可组合性。

官网：http://reactivex.io/rxjs/manual/overview.html#anatomy-of-an-observable