RxJS核心解析——Observer & Subscription

前面我们谈到一个数据流的生命周期分为创建，订阅，执行，完成四个部分。在创建时，我们传入了一个订阅函数作为订阅时数据流出的逻辑。这一节我们就来具体谈谈执行订阅逻辑时涉及到的概念——Observer和Subscription。

Observer

现在我们知道了Observable是数据源，当我们订阅它时会执行其中订阅函数的逻辑流出数据，那么这个数据会流到哪呢？也许你已经注意到了，订阅函数中传入了一个参数——Observer（观察者）,每个流出数据的方法都由它来执行。这就是数据流推送数据的消费者。

之前我们谈到数据流会流出3种类型的数据。相应的，观察者也必然需要3种处理数据的逻辑。一个完整的观察者定义如下：

let observer = {
  next: x => console.log('value: x'),
  error: err => console.log('error', err),
  complete: () => console.log('over')
};

因为数据流会在订阅时流出，所以我们只需要在订阅时传入观察者即可连接这两者来处理数据了。例子如下：

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

// 传入上文定义的观察者
observable.subscribe(observer);
// 输出结果依次为：'value:1','value:2','over'

需要注意的是，传入observable是最完备的订阅方法，但我们也有简化的订阅方式。

1. observer不定义3种逻辑也是可行的，当接收到未定义逻辑的数据时会忽略该数据。
2. 可以用回调函数的方式代替新建一个observer对象再传入的方法，3个参数依次代表正常，错误，完成时的处理逻辑。

Subscription

我们之前也提到过，数据流的执行有可能是无限的，我们需要特殊的方法来控制数据流的流动时机。既然订阅数据流时就会开启流动，那么很自然的就想到终止数据流的开关也应该同时机产生，这就是订阅时返回的对象——Subscription。这个对象通过unsubscribe方法截断数据流。并且该对象可以通过add方法嵌套进行多个数据流的截断控制，remove方法撤销嵌套的Subscription。实例如下：

let ob1 = Rx.Observable.interval(400);
let ob2 = Rx.Observable.interval(300);

let subscription1 = ob1.subscribe(x => console.log('first', x));
let subscription2 = ob2.subscribe(x => console.log('second', x));

subscription.add(subscription2);

setTimeout(() => {
  subscription.unsubscribe();
}, 1000);

/* 输出结果为：
 second: 0
 first: 0
 second: 1
 first: 1
 second: 2 */

总结

综上可以看出，Observer和Subscription在数据流动的过程中起作用。前者控制接收数据后的处理逻辑，后者控制数据流动的结束时机。