subscribe 方法内可观察取消订阅

Question

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

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

例子：

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

Answer 1

问题

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

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

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

隐藏的约束

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

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

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

解决方案

1. 保持脆弱的代码库

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

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

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

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

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

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

以下是您可以如何使用上面的模式，而不必担心可观察对象正在使用哪个调度程序。这实际上与调度程序无关，尽管它可能使相当简单的任务变得比它需要的复杂得多。

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();