rxjs - 通过主题订阅停止可观察

Question

我创建了一个 Observable 生成可能无限量的数据(例如计时器)。此数据通过主题访问，因此多个观察者将收到相同的值。

如何停止 Observable 产生新的值？ (不修改 Observable 的实现)

// custom Observable, to visualize internal behavior
const timer$ = new rxjs.Observable(subscriber => {
  console.log("observable init");

  var counter = 0;
  const intervalId = setInterval(() => {
    ++counter;
    console.log("observable %s",counter);
    subscriber.next(counter);
  }, 1000);

  return () => {
    console.log("observable teardown");
    clearTimeout(intervalId);
  }
});

// subscribe through a subject
const subject$ = new rxjs.Subject();
timer$.subscribe(subject$);

const subscription = subject$.subscribe(value => console.log("observer %s", value));

// cancel subscription
setTimeout(() => {
  console.log("unsubscribe observer");
  subscription.unsubscribe();
  // TODO how to stop Observable generating new values?
}, 3000);

jsfiddle:https://jsfiddle.net/gy4tfd5w/

Answer 1

幸运的是，在 RxJS 中有一种专门而优雅的方法来解决这个问题。

你的要求是有

multiple observers [...] receive the same values

这称为多播 observable，并且有某些运算符用于从普通的“冷”observable 创建一个。

例如，不是直接创建 Subject 的实例，而是可以将可观察到的源通过管道传递给 share 运算符，它将创建 Subject 给你。 share 的文档如下:

Returns a new Observable that multicasts (shares) the original Observable. As long as there is at least one Subscriber this Observable will be subscribed and emitting data. When all subscribers have unsubscribed it will unsubscribe from the source Observable.

最后一句显示了 share 和 source$.subscribe(subject) 之间的细微差别。使用 share 时，一个所谓的 refCount 会保留下来，当没有订阅者离开时，它会自动从其来源取消订阅 Subject。

应用于您的代码，它看起来像这样:

const timer$ = 
    new rxjs.Observable(subscriber => {// your unchanged implementation})
    .pipe(
        share()
    );

const subscription = timer$.subscribe(value => console.log("observer %s", value));

这是包含您的示例代码的完整版本:

https://jsfiddle.net/50q746ad/

顺便说一下，share 并不是唯一执行多播的运算符。 There are great learning resources在该主题上更深入。