如何使用 takeUntil RxJS 操作符来声明性地管理订阅

简介

Angular 处理取消订阅可观察对象的操作，比如从 HTTP 服务返回的可观察对象或者使用 async 管道时。然而，对于其他情况，管理所有订阅并确保取消长期存在的订阅可能会变得困难。而且，取消大部分订阅的策略也会带来自己的问题。

在本文中，您将看到一个依赖于手动订阅和取消订阅的 Angular 应用示例。然后，您将比较它与使用 takeUntil 操作符来声明性地管理订阅的 Angular 应用示例。

先决条件

如果您想跟着本文学习，您需要：

对 RxJS 库有一定的了解，特别是 Observable 和 Subscription 将会有所帮助。
对 Apollo 和 GraphQL 有一定的了解会有所帮助，但不是必需的。

本教程经过 Node v15.3.0、npm v6.14.9、@angular/core v11.0.4、rxjs v6.6.3、apollo-angular v2.1.0、graph-tag v2.11.0 的验证。本文已经根据从早期版本的 @angular/core 和 rxjs 迁移的变化进行了编辑。

手动取消订阅

让我们从一个示例开始，您将在其中手动取消订阅两个订阅。

在这个示例中，代码正在订阅 Apollo 的 watchQuery 来从 GraphQL 端点获取数据。

该代码还创建了一个间隔可观察对象，当调用 onStartInterval 方法时，您将订阅该对象。

import { Component, OnInit, OnDestroy } from '@angular/core';import { Subscription, interval } from 'rxjs';import { Apollo } from 'apollo-angular';
import gql from 'graphql-tag';@Component({ ... })
export class AppComponent implements OnInit, OnDestroy {myQuerySubscription: Subscription;myIntervalSubscription: Subscription;constructor(private apollo: Apollo) {}ngOnInit() {this.myQuerySubscription = this.apollo.watchQuery<any>({query: gql`query getAllPosts {allPosts {titledescriptionpublishedAt}}`}).valueChanges.subscribe(({data}) => {console.log(data);});}onStartInterval() {this.myIntervalSubscription = interval(250).subscribe(value => {console.log('Current value:', value);});}ngOnDestroy() {this.myQuerySubscription.unsubscribe();if (this.myIntervalSubscription) {this.myIntervalSubscription.unsubscribe();}}
}

现在想象一下，您的组件有许多类似的订阅，当组件被销毁时，确保一切都被取消订阅可能会变得相当复杂。

使用 `takeUntil` 声明性地取消订阅

解决方案是使用 takeUntil 操作符来组合订阅，并使用一个在 ngOnDestroy 生命周期钩子中发出真值的主题。

以下代码片段执行了完全相同的操作，但这次代码将以声明性的方式取消订阅。您会注意到一个额外的好处是，您不再需要保留对我们订阅的引用。

import { Component, OnInit, OnDestroy } from '@angular/core';import { Subject, interval } from 'rxjs';
import { takeUntil } from 'rxjs/operators';import { Apollo } from 'apollo-angular';
import gql from 'graphql-tag';@Component({ ... })
export class AppComponent implements OnInit, OnDestroy {destroy$: Subject<boolean> = new Subject<boolean>();constructor(private apollo: Apollo) {}ngOnInit() {this.apollo.watchQuery<any>({query: gql`query getAllPosts {allPosts {titledescriptionpublishedAt}}`}).valueChanges.pipe(takeUntil(this.destroy$)).subscribe(({data}) => {console.log(data);});}onStartInterval() {interval(250).pipe(takeUntil(this.destroy$)).subscribe(value => {console.log('Current value:', value);});}ngOnDestroy() {this.destroy$.next(true);this.destroy$.unsubscribe();}
}