javascript - RxJs 5 share() 运算符如何工作？

Question

我并不是 100% 清楚 RxJs 5 share() 运算符是如何工作的，请参阅此处的 latest docs 。 Jsbin的问题here .

如果我创建一个包含一系列 0 到 2 的可观察量，每个值间隔一秒:

var source = Rx.Observable.interval(1000)
.take(5)
.do(function (x) {
    console.log('some side effect');
});

如果我为此 observable 创建两个订阅者:

source.subscribe((n) => console.log("subscriptor 1 = " + n));
source.subscribe((n) => console.log("subscriptor 2 = " + n));

我在控制台中得到这个:

"some side effect ..."
"subscriptor 1 = 0"
"some side effect ..."
"subscriptor 2 = 0"
"some side effect ..."
"subscriptor 1 = 1"
"some side effect ..."
"subscriptor 2 = 1"
"some side effect ..."
"subscriptor 1 = 2"
"some side effect ..."
"subscriptor 2 = 2"

我以为每个订阅都会订阅同一个Observable，但事实似乎并非如此!这就像订阅行为创建了一个完全独立的 Observable!

但是如果将 share() 运算符添加到源可观察对象中:

var source = Rx.Observable.interval(1000)
.take(3)
.do(function (x) {
    console.log('some side effect ...');
})
.share();

然后我们得到这个:

"some side effect ..."
"subscriptor 1 = 0"
"subscriptor 2 = 0"
"some side effect ..."
"subscriptor 1 = 1"
"subscriptor 2 = 1"
"some side effect ..."
"subscriptor 1 = 2"
"subscriptor 2 = 2"

这就是我在没有 share() 的情况下所期望的。

这里发生了什么，share()运算符如何工作？每个订阅都会创建一个新的 Observable 链吗？

Answer 1

请注意，您使用的是 RxJS v5，而您的文档链接似乎是 RxJS v4。我不记得具体细节，但我认为 share 运营商经历了一些变化，特别是在完成和重新订阅方面，但不要相信我的话。

回到你的问题，正如你在研究中所表明的，你的期望与库设计不符。 Observables 延迟实例化它们的数据流，具体是在订阅者订阅时启动数据流。当第二个订阅者订阅同一个可观察值时，另一个新的数据流将启动，就像它是第一个订阅者一样(所以是的，每个订阅都会创建一个新的可观察值链，正如您所说)。这就是 RxJS 术语中所创造的冷可观察值，也是 RxJS 可观察值的默认行为。如果您想要一个将数据发送给数据到达时的订阅者的可观察量，则这被称为热可观察量，获得热可观察量的一种方法是使用 share 运算符。

您可以在此处找到订阅和数据流的说明:Hot and Cold observables : are there 'hot' and 'cold' operators? (这对 RxJS v4 有效，但大部分对 v5 有效)。