Observable 与 Observer

我们先看到一个很普通的 Observable，它每秒发出一个数字，并且我们只需要前 6 个数字：

const observable = Rx.Observable.interval(1000).take(6);

const observerA = {
  next: x => console.log(`A next ${x}`),
  error: e => console.error(`A error ${e}`),
  complete: () => console.log('A done')
};

observable.subscribe(observerA);

/* 程序输出：
  "A next 0"
  "A next 1"
  "A next 2"
  "A next 3"
  "A next 4"
  "A next 5"
  "A done"
*/

查看例子

我们知道，当 observable.subscribe(observerA) 之后，observable 就会开始执行。如果我们希望在 observable 订阅了 observerA 之后，间隔一段时间后再订阅另一个观察者 observerB，即我们想要 observerble 能够广播这两个观察者：

const observable = Rx.Observable.interval(1000).take(6);

const observerA = {
  next: x => console.log(`A next ${x}`),
  error: e => console.error(`A error ${e}`),
  complete: () => console.log('A done')
};

observable.subscribe(observerA);

const observerB = {
  next: x => console.log(`B next ${x}`),
  error: e => console.error(`B error ${e}`),
  complete: () => console.log('B done')
};

setTimeout(() => {
  observable.subscribe(observerB);
}, 2000);

/* 程序输出：
  "A next 0"
  "A next 1"
  "A next 2"
  "B next 0"
  "A next 3"
  "B next 1"
  "A next 4"
  "B next 2"
  "A next 5"
  "A done"
  "B next 3"
  "B next 4"
  "B next 5"
  "B done"
 */

查看例子

此时，我们发现两个观察者 observerA 及 observerB 并没有获得一致的订阅反馈，比如，当值 2 到来时，我们希望两个观察者都能接收到 2，然而在这个例子中，observerA 收到 2 时，observerB 收到了 0。

出现该问题的原因即在于 subscribe 方法会驱动 Observable 从头开始执行：

observable: 0--1--2--3--4--5|
observerA:  0--1--2--3--4--5|
observerB:           0--1--2--3--4--5|

如果我们想要 observable 发出值的时候，两个观察者能获得一致响应，可以这样修改代码：

const observable = Rx.Observable.interval(1000).take(6);

const observerA = {
  next: x => {
    console.log(`A next ${x}`);
    observerB.next(x);
  },
  error: e => {
    console.error(`A error ${e}`);
    observerB.error(e);
  },
  complete: () => {
    console.log('A done');
    observerB.complete();
  }
};

observable.subscribe(observerA);

const observerB = {
  next: x => console.log(`B next ${x}`),
  error: e => console.error(`B error ${e}`),
  complete: () => console.log('B done')
};

/* 程序输出：
  "A next 0"
  "B next 0"
  "A next 1"
  "B next 1"
  "A next 2"
  "B next 2"
  "A next 3"
  "B next 3"
  "A next 4"
  "B next 4"
  "A next 5"
  "B next 5"
  "A done"
  "B done"
 */

查看例子