订阅方法中的可观察取消订阅

Question

我尝试在 subscribe 方法中取消订阅。好像可行，我在互联网上没有找到可以这样做的示例。

我知道取消订阅该方法或使用管道限制它还有许多其他可能性。请不要提出任何其他解决方案，但请回答为什么您不应该这样做或者这是一种可能的方式？

示例：

let localSubscription = someObservable.subscribe(result => {
  this.result = result;
  if (localSubscription && someStatement) {
    localSubscription.unsubscribe();
  }
});

Answer 1

这是工作方法，但 RxJS 主要推荐在 Angular 中使用async 管道。这是完美的解决方案。在您的示例中，您将 result 分配给对象属性，这不是一个好习惯。

如果您在模板中使用变量，则只需使用async 管道。如果您不这样做，只需以这种方式使其可观察：

private readonly result$ = someObservable.pipe(/...get exactly what you need here.../)

然后您可以在需要时使用您的结果$：在其他可观察对象或模板中。

您也可以使用pipe(take(1)) 或pipe(first()) 取消订阅。还有一些其他管道方法允许您取消订阅而无需额外的代码。

Answer 2

退订数据有多种方式：

Method 1: Unsubscribe after subscription; (Not preferred)

let localSubscription = someObservable.subscribe(result => {
  this.result = result;
}).unsubscribe();
---------------------
Method 2: If you want only first one or 2 values, use take operator or first operator

a) let localSubscription = 
  someObservable.pipe(take(1)).subscribe(result => {
     this.result = result;
  });

b) let localSubscription = 
  someObservable.pipe(first()).subscribe(result => {
     this.result = result;
  });
---------------------
Method 3: Use Subscription and unsubscribe in your ngOnDestroy();

let localSubscription = 
  someObservable.subscribe(result => {
     this.result = result;
  });

ngOnDestroy() { this.localSubscription.unsubscribe() }
----------------------

Method 4: Use Subject and takeUntil Operator and destroy in ngOnDestroy

let destroySubject: Subject<any> = new Subject();

let localSubscription = 
  someObservable.pipe(takeUntil(this.destroySubject)).subscribe(result => {
     this.result = result;
  });

ngOnDestroy() { 
  this.destroySubject.next();
  this.destroySubject.complete();
}

我个人更喜欢方法 4，因为如果您在一个页面中有多个订阅，则可以使用相同的销毁主题。

Answer 3

问题

您在上面使用的模式有时会起作用，有时则不会。这里有两个示例，您可以尝试自己运行它们。一个会抛出错误，另一个不会。

const subscription = of(1,2,3,4,5).pipe(
  tap(console.log)
).subscribe(v => {
  if(v === 4) subscription.unsubscribe();
});

输出：

1
2
3
4
Error: Cannot access 'subscription' before initialization

---

类似的东西：

const subscription = of(1,2,3,4,5).pipe(
  tap(console.log),
  delay(0)
).subscribe(v => {
  if (v === 4) subscription.unsubscribe();
});

输出：

1
2
3
4

这一次你没有收到错误，但你在源 observable of(1,2,3,4,5) 发出 5 之前也取消了订阅

隐藏的约束

如果您熟悉 RxJS 中的调度程序，您可能会立即发现额外的隐藏信息，这些信息允许一个示例工作而另一个不工作。

delay（即使是 0 毫秒的延迟）返回一个使用异步调度程序的 Observable。这实际上意味着当前代码块将在延迟的 observable 有机会发出之前完成执行。

这保证在单线程环境中（如当前浏览器中的 Javascript 运行时）您的订阅已被初始化。

解决方案

1。保留脆弱的代码库

一种可能的解决方案是忽略常识并继续使用这种模式来取消订阅。为此，您和您团队中可能使用您的代码作为参考或某天可能需要维护您的代码的任何人都必须承担额外的认知负担，即记住哪个 observable 使用了正确的调度程序。

更改应用程序某个部分中可观察对象转换数据的方式可能会导致依赖于异步调度程序提供的此数据的应用程序的每个部分出现意外错误。

例如：查询服务器时运行良好的代码可能会在同步返回兑现结果时中断。看起来像是优化的东西，现在会对你的代码库造成严重破坏。当出现此类错误时，可能很难追查到源头。

最后，如果浏览器（或者您在 Node.js 中运行代码）开始支持多线程环境，那么您的代码将不得不在没有这种增强的情况下勉强应付，或者被重新编写。

2。使“在订阅回调中取消订阅”成为一种安全模式

惯用的 RxJS 代码尽可能地尝试与调度无关。

这里是你可以如何使用上面的模式，而不用担心 observable 使用的是哪个调度器。这实际上与调度程序无关，尽管它可能使一个相当简单的任务比它需要的复杂得多。

const stream = publish()(of(1,2,3,4,5));

const subscription = stream.pipe(
  tap(console.log)
).subscribe(x => {
  if(x === 4) subscription.unsubscribe();
});

stream.connect();

这让您可以安全地使用“在订阅中取消订阅”模式。无论调度程序如何，这将始终有效，并且如果（例如）您将代码置于多线程环境中，它将继续有效（上面的 delay 示例可能会中断，但这不会）。

3。 RxJS 运算符

最好的解决方案是使用代表您处理订阅/取消订阅的运营商。在最好的情况下，它们不需要额外的认知负荷，并且在更奇特的情况下能够相对较好地控制/管理错误（远距离的怪异动作较少）。

大多数高阶运算符都这样做（concat、merge、concatMap、switchMap、mergeMap 等）。 take、takeUntil、takeWhile 等其他运算符让您可以使用更具声明性的样式来管理订阅。

在可能的情况下，这些更可取，因为它们都不太可能在使用它们的团队中引起奇怪的错误或混乱。

上面的例子重写了：

of(1,2,3,4,5).pipe(
  tap(console.log)
  first(v => v === 4)
).subscribe();