Rx.Subject 是一个热门的 observable 吗？

Question

代码

const a = new Rx.Subject().do(x => console.log('a'))
const b = a.mapTo(0)
const c = a.mapTo(1)
const d = Rx.Observable.merge(b, c)
d.subscribe(x => console.log('d'))
a.next(3)

还有输出

a
d
a
d

为什么 a 会打印两次？ Rx.Subject 不是一个热门的 observable 吗？

Answer 1

Subject 本身是热门/共享的。

但是：您附加的任何（大多数！）运算符都将创建一个新流，以先前的流（在本例中为 Subject）作为源 - 但是，新流是（对于大多数运营商而言）不热，只有通过附加 热运营商（如 share 或 publish 等...）派生热流才能使其变得热。

所以当你share 你的do 时，一切都应该按预期工作。

const a = new Rx.Subject().do(x => console.log('a')).share();
const b = a.mapTo(0);
const c = a.mapTo(1);
const d = Rx.Observable.merge(b, c)
d.subscribe(x => console.log('d'));
a.next(3);

<script src="https://unpkg.com/rxjs/bundles/Rx.min.js"></script>

Answer 2

您需要了解冷/热可观察对象和主题。

cold Observable 是一个 Observable，每次订阅时都会重新执行其订阅处理程序：

const cold = new Observable(function subscribe(observer) {
  console.log('subscribed');
  observer.next(Math.random());
  observer.complete();
});
// > subscribed
// sub 1: 0.1231231231231
cold.subscribe((num) => console.log('sub 1:', num));
// > subscribed
// sub 2: 0.09805969045
cold.subscribe((num) => console.log('sub 2:', num));

热 Observable 是源 Observable（冷或其他），在源和订阅者之间有一个主题。当订阅一个热的 Observable 时，订阅会在内部透明地路由到内部的 Subject，Subject 会订阅到源 Observable。这确保了源 Observable 只有一个订阅者（Subject），并且 Subject 与许多订阅者共享源的值：

const cold = new Observable(function subscribe(observer) {
  console.log('subscribed');
  observer.next(Math.random());
  observer.complete();
});

const hot = cold.publish();
hot.subscribe((num) => console.log('sub 1:', num));
hot.subscribe((num) => console.log('sub 2:', num));
hot.connect(); // <-- this subscribes the inner Subject to the cold source
// > subscribed
// > sub 1: 0.249848935489
// > sub 2: 0.249848935489

您可以通过多播使 Observable 成为热的，它采用一个函数，该函数返回一个主题，以便在连接时使用。为方便起见，也有创建特定类型主题的多播变体（例如发布）。 publish() 是multicast(() => new Subject()) 的便捷方法

除了connect()，它将内部Subject订阅源并返回底层Subscription，你可以调用refCount()，它返回一个Observable。当refCount()返回的Observable被订阅一次后，内部会自动调用connect()，后续订阅不会重连。当所有订阅者都取消订阅时，refCount 将自动从源取消订阅内部主题。 share() 是source.publish().refCount() 的便捷方法。

所以，它会起作用的，

const a = new Rx.Subject().do(x => console.log('a')).share();
const b = a.mapTo(0);
const c = a.mapTo(1);
const d = Rx.Observable.merge(b, c)
d.subscribe(x => console.log('d'));
a.next(3);