热&冷的 Observables

Observable 有冷热两种类型。我们先来看看什么是冷的 observable 。如果是冷的 observable 的话，那么两个订阅者得到值是两份完全相同的副本，示例如下：

  1. // 冷的 observable 示例
  2. let stream$ = Rx.Observable.of(1,2,3);
  3. //订阅者 1: 1,2,3
  4. stream.subscribe(
  5.    data => console.log(data),
  6.    err => console.error(err),
  7.    () => console.log('completed')
  8. )
  10. //订阅者 2: 1,2,3
  11. stream.subscribe(
  12.    data => console.log(data),
  13.    err => console.error(err),
  14.    () => console.log('completed')
  15. )

如果是热的 observable 的话，订阅者只能收到当它开始订阅后的值，这很像是足球比赛的实况直播，如果你在开场5分钟后才开始观看，你会错失开场前5分钟的一切，从观看的这一刻起你才开始接收数据：

  1. let liveStreaming$ = Rx.Observable.interval(1000).take(5);
  3. liveStreaming$.subscribe(
  4.   data => console.log('subscriber from first minute')
  5.   err => console.log(err),
  6.   () => console.log('completed')
  7. )
  9. setTimeout(() => {
  10.    liveStreaming$.subscribe(
  11.   data => console.log('subscriber from 2nd minute')
  12.   err => console.log(err),
  13.   () => console.log('completed')
  14. )
  15. },2000)

由冷及热 - 凯蒂·佩里模式

上面的示例其实并不是真正的热的 observable，事实上两个订阅者接收到的值都是0,1,2,3,4。因为这是一场足球比赛的实况直播，所以这样的结果并不是我们想要的，那么如何来修复呢？

需要两个部件来将冷的 observable 转变成热的， publish() 和 connect() 。

  1. let publisher$ = Rx.Observable
  2. .interval(1000)
  3. .take(5)
  4. .publish();
  7. publisher$.subscribe(
  8.   data => console.log('subscriber from first minute',data),
  9.   err => console.log(err),
  10.   () => console.log('completed')
  11. )
  13. setTimeout(() => {
  14.     publisher$.subscribe(
  15.         data => console.log('subscriber from 2nd minute', data),
  16.         err => console.log(err),
  17.         () => console.log('completed')
  18.     )
  19. }, 3000)
  22. publisher$.connect();

在这个案例中，我们看到第一个订阅者输出的是0,1,2,3,4，而第二个输出的是3,4。很明显订阅的时间点是很重要的。

暖的 Observables

这是 observalbes 的另外一种类型，它的表现很像热的 observable ，但它在某种程度上是惰性的。我想表达的是从本质上来说，在有订阅发生之前它们不会发出任何值。让我们来比较一下热的和暖的 observable

热的 observable

  1. let stream$ = Rx.Observable
  2. .interval(1000)
  3. .take(4)
  4. .publish();
  6. stream$.connect();
  8. setTimeout(() => {
  9.   stream$.subscribe(data => console.log(data))
  10. }, 2000);

这里我们可以看到热的 observable 会丢失第一个发出的值，因为订阅是延迟发生的。

和暖的 observable 进行下对比

暖的 observable

  1. let obs = Rx.Observable.interval(1000).take(3).publish().refCount();
  3. setTimeout(() => {
  4.     obs.subscribe(data => console.log('sub1', data));
  5. },1100)
  7. setTimeout(() => {
  8.     obs.subscribe(data => console.log('sub2', data));
  9. },2100)

refCount() 操作符确保 observable 变成暖的，也就是不会发出值直到 sub1 订阅了流。另一方面 sub2 是后加入的，也就是说订阅接收的是当前的值，而无法接收订阅之前的值。

所以这个输出是：

  1. sub1 : 0
  2. sub1 : 1
  3. sub2 : 1
  4. sub1 : 2
  5. sub2 : 2

如上面显示的，第一个订阅者是从0开始。如果它是热的话，它会从一个更大的数字开始，因为它是后加入的。当第二个订阅者达到时，它得到的第一个数字不是0而是1，这也说明了它确实变热了。

天生的热 observables

通常来说，如果 observable 的值被立即发出而不需要订阅者的话，那么就认为它是热的，一个最常见的例子就是 mousemove 。其它大多数热的 observables 都是通过使用 publish() 和 connect() ，或者使用 share() 操作符将冷的 observables 转变成热的结果。

共享

共享意味着要使用一个十分有用，叫做 share() 的操作符。想象一下你有这样一个普通的冷 observable：

  1. let stream$ = Rx.Observable.create((observer) => {
  2.     observer.next( 1 );
  3.     observer.next( 2 );
  4.     observer.next( 3 );
  5.     observer.complete();
  6. }).share()
  8. stream$.subscribe(
  9.     (data) => console.log('subscriber 1', data),
  10.     err => console.error(err),
  11.     () => console.log('completed')
  12. );
  13. stream$.subscribe(
  14.     (data) => console.log('subscriber 2', data),
  15.     err => console.error(err),
  16.     () => console.log('completed')
  17. );

如果在 observer.next(1) 打个断点，你会注意到它执行了两次，每个订阅者一次。这个行为是我们对冷的 observable 的期望。共享操作符用了一种不同的方式将其转换成热的 observable，事实上，它不仅在正确的条件下转变成热的 observable，而且在某些条件下可以回退成冷的 observable 。那么这些条件是？

1) 创建热的 Observable： 当有新的订阅时 Observable 还未完成并且订阅者数量大于0

2) 创建冷的 Observable 在新的订阅之前订阅者的数量已经变成了0，也就是说，一个或多个订阅存在过一段时间，但是在新的订阅发生前已经取消了订阅。

3) 创建冷的 Observable 当新的订阅发生之前 Observable 已经完成

结果就是一个活动的 Observable 要继续产生值至少要有一个存在的订阅者。我们可以看到情况1)中的 Observable 在第有两个订阅者之前是休眠的，当订阅发生时它会立即转变成热的从而开始共享数据。