如何在 subscribe 方法中取消订阅 RXJS 订阅？答案

【问题标题】：how to unsubscribe a RXJS subscription inside the subscribe method?如何在 subscribe 方法中取消订阅 RXJS 订阅？
【发布时间】：2021-01-28 17:02:05
【问题描述】：

我有一些 javascript：

this.mySubscription = someObservable.subscribe((obs: any) => {
   this.mySubscription.unsubscribe();
   this.mySubscription = undefined;
}

在执行时，控制台记录错误ERROR TypeError: Cannot read property 'unsubscribe' of undefined。我想知道为什么我不能在 subscribe lambda 函数中取消订阅。有正确的方法吗？我已经阅读了一些关于使用虚拟主题并完成它们或使用 takeUntil/takeWhile 和其他管道运算符 workArounds 的内容。

在订阅的订阅功能中取消订阅的正确方法/解决方法是什么？

我目前正在使用这样的虚拟订阅：

mySubscription: BehaviorSubject<any> = new BehaviorSubject<any>(undefined);


// when I do the subscription:
dummySubscription: BehaviorSubject<any> = new BehaviourSubject<any>(this.mySubscription.getValue());
this.mySubscription = someObservable.subscribe((obs: any) => {
    // any work...
    dummySubscription.next(obs);
    dummySubscription.complete();
    dummySubscription = undefined;
}, error => {
    dummySubscription.error(error);
});

dummySubscription.subscribe((obs: any) => {
    // here the actual work to do when mySubscription  emits a value, before it should have been unsubscribed upon
}, err => {
    // if errors need be
});

【问题讨论】：

标签： rxjs rxjs-observables

【解决方案1】：

您不应尝试在 subscribe 函数中取消订阅。
您可以使用 take、takeWhile 或 takeUntil 等运算符取消订阅。

采取

在someObservable 发出n 次后，使用take(n) 取消订阅。

someObservable.pipe(
  take(1)
).subscribe(value => console.log(value));

采取当

当发出的值不符合条件时，使用takeWhile 取消订阅。

someObservable.pipe(
  takeWhile(value => valueIsSave(value))
).subscribe(value => console.log(value));

valueIsSave(value): boolean {
  // return true if the subscription should continue
  // return false if you want to unsubscribe on that value
}

采取直到

当可观察的obs$ 发出时，使用takeUntil(obs$) 取消订阅。

const terminate = new Subject();

someObservable.pipe(
  takeUntil(terminate)
).subscribe(value => console.log(value));

unsub() { 
  terminate.next() // trigger unsubscribe
}

【讨论】：

何时取消订阅的条件很复杂，取决于发出的值。
@MojioMS 听起来takeWhile 应该适合你
在测试不同的 rxjs 管道运算符时，我因意外订阅多次而感到困惑。然而，takeWhile 做了我想要的。

【解决方案2】：

如果你让你的流异步，你正在做的应该工作。例如，这将不起作用：

const sub = from([1,2,3,4,5,6,7,8,9,10]).subscribe(val => {
  console.log(val);
  if(val > 5) sub.unsubscribe();
});

但这会起作用：

const sub2 = from([1,2,3,4,5,6,7,8,9,10]).pipe(
  delay(0)
).subscribe(val => {
  console.log(val);
  if(val > 5) sub2.unsubscribe();
});

因为 JS 事件循环是相当可预测的（代码块总是运行到完成），如果您的流的任何部分是异步的，那么您可以确定您的订阅将在调用 lambda 回调之前定义。

你应该这样做吗？

可能不会。如果您的代码依赖于您的语言/编译器/解释器/等的内部（否则隐藏）机制，那么您创建了脆弱的代码和/或难以维护的代码。下一个查看我的代码的开发人员将会对为什么会有 delay(0) 感到困惑 - 看起来它不应该做任何事情。

请注意，在subscribe() 中，您的 lambda 可以访问其闭包以及当前的流变量。 takeWhile() 运算符可以访问相同的闭包和相同的流变量。

from([1,2,3,4,5,6,7,8,9,10]).pipe(
  takeWhile(val => {
    // add custom logic
    return val <= 5;
  })
).subscribe(val => {
  console.log(val);
});

takeWhile() 可以转换为sub = subscribe(... sub.unsubscibe() ... ) 的任何内容，并且具有不需要您管理订阅对象并且更易于阅读/维护的额外好处。

【讨论】：