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

多播 Observable 在底层是通过使用 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 执行。

引用计数

手动调用 connect() 并处理 Subscription 通常太笨重。通常，当第一个观察者到达时我们想要自动地连接，而当最后一个观察者取消订阅时我们想要自动地取消共享执行。

请考虑以下示例，下面的列表概述了 Subscriptions 发生的经过：

第一个观察者订阅了多播 Observable

多播 Observable 已连接

next 值 0 发送给第一个观察者

第二个观察者订阅了多播 Observable

next 值 1 发送给第一个观察者

next 值 1 发送给第二个观察者

第一个观察者取消了多播 Observable 的订阅

next 值 2 发送给第二个观察者

第二个观察者取消了多播 Observable 的订阅

多播 Observable 的连接已中断(底层进行的操作是取消订阅)

要实现这点，需要显式地调用 connect()，代码如下：

var source = Rx.Observable.interval(500);

var subject = new Rx.Subject();

var multicasted = source.multicast(subject);

var subscription1, subscription2, subscriptionConnect;

subscription1 = multicasted.subscribe({

  next: (v) => console.log('observerA: ' + v)

});

// 这里我们应该调用 `connect()`，因为 `multicasted` 的第一个

// 订阅者关心消费值

subscriptionConnect = multicasted.connect();

setTimeout(() => {

  subscription2 = multicasted.subscribe({

    next: (v) => console.log('observerB: ' + v)

  });

}, 600);

setTimeout(() => {

  subscription1.unsubscribe();

}, 1200);

// 这里我们应该取消共享的 Observable 执行的订阅，

// 因为此后 `multicasted` 将不再有订阅者

setTimeout(() => {

  subscription2.unsubscribe();

  subscriptionConnect.unsubscribe(); // 用于共享的 Observable 执行

}, 2000);

复制代码

如果不想显式调用 connect()，我们可以使用 ConnectableObservable 的 refCount() 方法(引用计数)，这个方法返回 Observable，这个 Observable 会追踪有多少个订阅者。当订阅者的数量从0变成1，它会调用 connect() 以开启共享的执行。当订阅者数量从1变成0时，它会完全取消订阅，停止进一步的执行。

refCount 的作用是，当有第一个订阅者时，多播 Observable 会自动地启动执行，而当最后一个订阅者离开时，多播 Observable 会自动地停止执行。

var source = Rx.Observable.interval(500);

var subject = new Rx.Subject();

var refCounted = source.multicast(subject).refCount();

var subscription1, subscription2, subscriptionConnect;

// 这里其实调用了 `connect()`，

// 因为 `refCounted` 有了第一个订阅者

console.log('observerA subscribed');

subscription1 = refCounted.subscribe({

  next: (v) => console.log('observerA: ' + v)

});

setTimeout(() => {

  console.log('observerB subscribed');

  subscription2 = refCounted.subscribe({

    next: (v) => console.log('observerB: ' + v)

  });

}, 600);

setTimeout(() => {

  console.log('observerA unsubscribed');

  subscription1.unsubscribe();

}, 1200);

// 这里共享的 Observable 执行会停止，

// 因为此后 `refCounted` 将不再有订阅者

setTimeout(() => {

  console.log('observerB unsubscribed');

  subscription2.unsubscribe();

}, 2000);

// 执行结果：

observerA subscribed

observerA: 0

observerB subscribed

observerA: 1

observerB: 1

observerA unsubscribed

observerB: 2

observerB unsubscribed

复制代码

refCount() 只存在于 ConnectableObservable，它返回的是 Observable，而不是另一个 ConnectableObservable 。

作者：chimyking

链接：https://juejin.im/post/5b46f08de51d45199358838a

来源：掘金

著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。

var subject = new Rx.ReplaySubject(3); // 为新的订阅者缓冲3个值

subject.subscribe({

  next: (v) => console.log('observerA: ' + v)

});

subject.next(1);

subject.next(2);

subject.next(3);

subject.next(4);

subject.subscribe({

  next: (v) => console.log('observerB: ' + v)

});

subject.next(5); 

作者：chimyking

链接：https://juejin.im/post/5b46f08de51d45199358838a

来源：掘金

著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。

var subject = new Rx.ReplaySubject(100, 500 /* windowTime */);

subject.subscribe({

  next: (v) => console.log('observerA: ' + v)

});

var i = 1;

setInterval(() => subject.next(i++), 200);

setTimeout(() => {

  subject.subscribe({

    next: (v) => console.log('observerB: ' + v)

  });

}, 1000);

// 输出:

observerA: 1

observerA: 2

observerA: 3

observerA: 4

observerA: 5

observerB: 3

observerB: 4

observerB: 5

observerA: 6

observerB: 6

...

作者：chimyking

链接：https://juejin.im/post/5b46f08de51d45199358838a

来源：掘金

著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。