Angular2 Observable 可观察对象

  可观察对象支持在应用中的发布者和订阅者之间传递消息。在需要进行事件处理,异步编程和处理多值的时候,可观察对象相对其他技术有显著的优点。

  可观察对象是声明式的 —— 也就是说,虽然你定义了一个用于发布值的函数,但是在有消费者订阅它之前,这个函数并不会实际执行。 订阅之后,当这个函数执行完或取消订阅时,订阅者就会收到通知。

创建 Observable(可观察对象)

  Rx.Observable.createObservable 构造函数的别名,它接收一个参数:subscribe 函数。

  使用 Observable 构造函数可以创建任何类型的可观察流。 当执行可观察对象的 subscribe() 方法时,这个构造函数就会把它接收到的参数作为订阅函数来运行。 订阅函数会接收一个 Observer 对象,并把值发布给观察者的 next() 方法。

如下:创建一个Observable,并每隔一秒向观察者发送‘hi’:

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

注意:Observables 可以使用 create 来创建, 但通常我们使用所谓的创建操作符, 像 offrominterval、等等。

如下:

Observable.of(...items) —— 返回一个 Observable 实例,它用同步的方式把参数中提供的这些值发送出来。
Observable.from(iterable) —— 把它的参数转换成一个 Observable 实例。 该方法通常用于把一个数组转换成一个(发送多个值的)可观察对象。

执行Observable

Observable 执行可以传递三种类型的值:

  • "next" :用来处理每个送达值。在开始执行后可能执行零次或多次。比如数字、字符串、对象,等等。
  • "error" :可选。用来处理错误通知,发送一个 JavaScript 错误 或 异常。错误会中断这个可观察对象实例的执行过程 。
  • "complete" :用来处理执行完毕(complete)通知。

  "Next" 通知是最重要,也是最常见的类型:它们表示传递给观察者的实际数据。

  "Error" 和 "Complete" 通知可能只会在 Observable 执行期间发生一次,并且只会执行其中的一个,一旦发送,那么之后不会再发送任何通知了。

下面是 Observable 执行的示例,它发送了三个 "Next" 通知,然后是 "Complete" 通知:

//用try..catch 捕获异常
var observable = Rx.Observable.create(function subscribe(observer) { try { observer.next(1); observer.next(2); observer.next(3); observer.complete();
observer.next(4); // 因为违反规约,所以不会发送
} catch (err) { observer.error(err); // 如果捕获到异常会发送一个错误 } });

清理 Observable 执行

  因为 Observable 执行可能会是无限的,并且观察者通常希望能在有限的时间内中止执行,所以我们需要一个 API 来取消执行。因为每个执行都是其对应观察者专属的,一旦观察者完成接收值,它必须要一种方法来停止执行,以避免浪费计算能力或内存资源。

  当你订阅了 Observable,你会得到一个 Subscription ,它表示进行中的执行。只要调用 unsubscribe() 方法就可以取消执行。

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

  当我们使用 create() 方法创建 Observable 时,Observable 必须定义如何清理执行的资源。你可以通过在 function subscribe() 中返回一个自定义的 unsubscribe 函数。

var observable = Rx.Observable.create(function subscribe(observer) {
  // 追踪 interval 资源
  var intervalID = setInterval(() => {
    observer.next('hi');
  }, 1000);

  // 提供取消和清理 interval 资源的方法
  return function unsubscribe() {
    clearInterval(intervalID);
  };
});

//这里返回一个函数声明
var subscription=observable.subscribe((x)=>console.log(x));
//执行取消订阅
subscription.unsubscribe();

  观察者是由 Observable 发送的值的消费者。观察者只是一组回调函数的集合,每个回调函数对应一种 Observable 发送的通知类型:nexterrorcomplete 。下面的示例是一个典型的观察者对象:

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'),
};

observable.subscribe(observer);

//观察者只是有三个回调函数的对象,每个回调函数对应一种 Observable 发送的通知类型。

Subscription (订阅)

  Subscription 基本上只有一个 unsubscribe() 函数,它不需要任何参数,这个函数用来释放资源或去取消 Observable 执行占用的资源。

  Subscription 还可以合在一起,这样一个 Subscription 调用 unsubscribe() 方法,可能会有多个 Subscription 取消订阅 。你可以通过把一个 Subscription 添加到另一个上面来做这件事:

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 和 childSubscription 都会取消订阅
  subscription.unsubscribe();
}, 1000);

  执行时,我们在控制台中看到:

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

  Subscriptions 还有一个 remove(otherSubscription) 方法,用来撤销一个已添加的子 Subscription 。

Subject (主体)

  Subject 是一种特殊类型的 Observable,它允许将值多播给多个观察者,所以 Subject 是多播的,而普通的 Observables 是单播的(每个已订阅的观察者都拥有 Observable 的独立执行)。

  每个 Subject 都是 Observable 。 - 对于 Subject,你可以提供一个观察者并使用 subscribe 方法,就可以开始正常接收值。从观察者的角度而言,它无法判断 Observable 执行是来自普通的 Observable 还是 Subject 。

  在 Subject 的内部,subscribe 不会调用发送值的新执行。它只是将给定的观察者注册到观察者列表中,类似于其他库或语言中的 addListener的工作方式。

  每个 Subject 都是观察者。 - Subject 是一个有如下方法的对象: next(v)error(e)complete() 。要给 Subject 提供新值,只要调用 next(theValue),它会将值多播给已注册监听该 Subject 的观察者们。

  在下面的示例中,我们为 Subject 添加了两个观察者,然后给 Subject 提供一些值:

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);

  下面是控制台的输出:

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

  因为 Subject 是观察者,这也就在意味着你可以把 Subject 作为参数传给任何 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 进行订阅

  执行结果:

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

  使用上面的方法,我们基本上只是通过 Subject 将单播的 Observable 执行转换为多播的。这也说明了 Subjects 是将任意 Observable 执行共享给多个观察者的唯一方式。

Subject的三种变体类型:

  • BehaviorSubject :主要用来表示随着时间的流逝,值会改变的情况。比如生日是Subject,而生日就是BehaviorSubject。
  • ReplaySubject :当有新的订阅者订阅时,可以指定回放的值得个数,或者以 window time (以毫秒为单位) 规定范围内发生的值。
  • AsyncSubject : 当被观察者的值推送完成时,才将最后一个值发送给观察者。类似于last ()操作符。

多播的 Observables

  多播 Observable 在底层是通过使用 Subject 使得多个观察者可以看见同一个 Observable 执行。

  多播 Observable” 通过 Subject 来发送通知,这个 Subject 可能有多个订阅者,然而普通的 “单播 Observable” 只发送通知给单个观察者。

  在底层,这就是 multicast 操作符的工作原理:观察者订阅一个基础的 Subject,然后 Subject 订阅源 Observable 。下面的示例与前面使用 observable.subscribe(subject) 的示例类似:

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

// 在底层使用了 `subject.subscribe({...})`:
multicasted.subscribe({
  next: (v) => console.log('observerA: ' + v)
});
multicasted.subscribe({
  next: (v) => console.log('observerB: ' + v)
});

// 在底层使用了 `source.subscribe(subject)`:
multicasted.connect();

  multicast 操作符返回一个 Observable,它看起来和普通的 Observable 没什么区别,但当订阅时就像是 Subject 。multicast 返回的是 ConnectableObservable,它只是一个有 connect() 方法的 Observable 。

  connect() 方法十分重要,它决定了何时启动共享的 Observable 执行。因为 connect() 方法在底层执行了 source.subscribe(subject),所以它返回的是 Subscription,你可以取消订阅以取消共享的 Observable 执行。