rxjs Observable：处理所有订阅的取消订阅答案

【问题标题】：rxjs Observable: handle unsubscribe of all subscriberxjs Observable：处理所有订阅的取消订阅
【发布时间】：2018-10-26 13:11:10
【问题描述】：

我有一个返回 Observable 的方法

subFoo(id): Observable<number> {
    return new Observable<number>(observer => {
        setTimeout(() => {
            observer.next(id);
        }, 1000);
    });
}

现在我订阅了 3 次，5 秒后全部退订：

const sub1 = subFoo(1).subscribe(result => console.log(result));
const sub2 = subFoo(2).subscribe(result => console.log(result));
const sub3 = subFoo(3).subscribe(result => console.log(result));

setTimeout(() => {
  sub1.unsubscribe();
  sub2.unsubscribe();
  sub3.unsubscribe();
}, 5000);

我可以处理所有监听器的完全取消订阅吗？

例如。（伪代码）：

subFoo(id): Observable<number> {
    return new Observable<number>(observer => {

        // something like this
        observer.onAllListenerAreUnsubscribed(() => {
           console.log('All Listener Are Unsubscribed!');
        });

        setTimeout(() => {
            observer.next(id);
        }, 1000);
    });
}

现场演示：https://stackblitz.com/edit/angular-ayl12r

【问题讨论】：

标签： javascript rxjs

【解决方案1】：

Observable 可以知道对其链的订阅。如果您想知道某人订阅了多少次，您可以自己计算：

let subscriptions = 0;

subFoo(id): Observable<number> {
  return new Observable<number>(observer => {
    subscriptions++;
    ...
    return (() => {
      if (--subscriptions === 0) {
        // do whatever...
      }
      ...
    })
  })
})

您还可以将“观察者端”的所有订阅收集到单个订阅中，然后在取消订阅时添加自定义处理程序：

const subs = new Subscription();
subs.add(subFoo(1).subscribe(...));
subs.add(subFoo(2).subscribe(...));
subs.add(subFoo(3).subscribe(...));
subs.add(() => {
  // do whatever...
});

subs.unsubscribe(); // Will unsubscribe all subscriptions and then call your custom method.

【讨论】：

【解决方案2】：

您可以在一行中取消订阅所有侦听器，因此无需处理该事件

subscriptions.add(sub1).add(sub2).add(sub3);

// Then unsubscribe all of them with a single 
subscriptions.unsubscribe();

【讨论】：

【解决方案3】：

通过一次完成所有可观察项，您可以确定不会发生任何数据泄漏。您可以创建一个在可观察对象停止发射时将发射的主题，并在您的可观察对象上使用 takeUntil() 运算符，如下所示：

const completeSubscription: Subject<void> = new Subject();

const sub1 = subFoo(1)
  .pipe(takeUntil(completeSubscription))
  .subscribe(result => console.log(result));
const sub2 = subFoo(2)
  .pipe(takeUntil(completeSubscription))
  .subscribe(result => console.log(result));
const sub3 = subFoo(3)
  .pipe(takeUntil(completeSubscription))
  .subscribe(result => console.log(result));

setTimeout(() => {
  completeSubscription.next();
  completeSubscription.complete();
}, 5000);

【讨论】：