Angular/RxJS 我什么时候应该取消订阅“订阅”

TL;博士

对于这个问题，有两种 Observables - 有限值和无限值。

http Observable 产生 finite (1) 值，类似 DOM 事件侦听器 Observable 产生 infinite 值。

如果您手动调用 subscribe（不使用异步管道），则从 infinite Observables 调用 unsubscribe。

不用担心有限的，RxJs 会处理它们的。

资料来源：

我在这里从 Angular 的 Gitter 中找到了 Rob Wormald 的答案。他说（为了清楚起见，我进行了重新组织，重点是我的）：如果它是单值序列（如 http 请求），则不需要手动清理（假设您手动订阅控制器）我应该说“如果它是一个序列完成”（其中单个值序列，一个 la http，是一个）如果它是一个无限序列，你应该取消订阅异步管道为你做的事情他还在这个 YouTube 视频中提到 Observables 说“他们自己清理...... .” 在完成的 Observables 的上下文中（比如 Promises，它总是完成，因为它们总是产生一个值并结束 - 我们从不担心取消订阅 Promises 以确保它们清理 XHR 事件侦听器，对吧？）同样在 Rangle Angular 2 指南在大多数情况下，我们不需要显式调用 unsubscribe 方法，除非我们想提前取消或者我们的 Observable 的生命周期比我们的订阅更长。 Observable 操作符的默认行为是在 .complete() 或 .error() 消息发布后立即处理订阅。请记住，RxJS 被设计为大多数时候以“即发即弃”的方式使用。 “我们的 Observable 的生命周期比我们的订阅更长”这句话何时适用？它适用于在 Observable 完成之前（或不久之前）销毁的组件内创建订阅。我认为这意味着如果我们订阅了一个 http 请求或一个发出 10 个值的 Observable，并且我们的组件在该 http 请求返回或发出 10 个值之前被销毁，我们仍然可以！当请求返回或最终发出第 10 个值时，Observable 将完成并清理所有资源。如果我们从同一个 Rangle 指南中查看这个示例，我们可以看到对 route.params 的订阅确实需要 unsubscribe()，因为我们不知道这些参数何时会停止更改（发出新值）。该组件可能会通过导航而被销毁，在这种情况下，路由参数可能仍会发生变化（从技术上讲，它们可能会在应用程序结束之前发生变化）并且订阅中分配的资源仍将被分配，因为尚未完成。在 NgEurope 的这段视频中，Rob Wormald 还说您不需要取消订阅 Router Observables。他还在 2016 年 11 月的这段视频中提到了 http 服务和 ActivatedRoute.params。Angular 教程的路由章节现在声明如下：路由器管理它提供的可观察对象并本地化订阅。当组件被销毁时，订阅会被清理，防止内存泄漏，所以我们不需要从路由参数 Observable 中取消订阅。 Here's a Discussion on the GitHub Issues for the Angular docs about Router Observables 其中 Ward Bell 提到对所有这些的澄清正在进行中。

我在 NGConf 上就这个问题与 Ward Bell 进行了交谈（我什至向他展示了这个他说是正确的答案），但他告诉我 Angular 的文档团队有一个未发布的解决方案（尽管他们正在努力让它获得批准） ）。他还告诉我，我可以用即将发布的官方推荐来更新我的 SO 答案。

今后我们都应该使用的解决方案是向所有在其类代码中对 Observable 进行 .subscribe() 调用的组件添加一个 private ngUnsubscribe = new Subject<void>(); 字段。

然后我们在 ngOnDestroy() 方法中调用 this.ngUnsubscribe.next(); this.ngUnsubscribe.complete();。

秘诀（如 @metamaker 所述）是在我们的每个 .subscribe() 调用之前调用 takeUntil(this.ngUnsubscribe)，这将保证在组件被销毁时所有订阅都将被清理。

例子：

import { Component, OnDestroy, OnInit } from '@angular/core';
// RxJs 6.x+ import paths
import { filter, startWith, takeUntil } from 'rxjs/operators';
import { Subject } from 'rxjs';
import { BookService } from '../books.service';

@Component({
    selector: 'app-books',
    templateUrl: './books.component.html'
})
export class BooksComponent implements OnDestroy, OnInit {
    private ngUnsubscribe = new Subject<void>();

    constructor(private booksService: BookService) { }

    ngOnInit() {
        this.booksService.getBooks()
            .pipe(
               startWith([]),
               filter(books => books.length > 0),
               takeUntil(this.ngUnsubscribe)
            )
            .subscribe(books => console.log(books));

        this.booksService.getArchivedBooks()
            .pipe(takeUntil(this.ngUnsubscribe))
            .subscribe(archivedBooks => console.log(archivedBooks));
    }

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

注意：将 takeUntil 运算符添加为最后一个很重要，以防止在运算符链中的中间 Observable 泄漏。

最近，在 Adventures in Angular 的一集中，Ben Lesh 和 Ward Bell 讨论了有关如何/何时取消订阅组件的问题。讨论从大约 1:05:30 开始。

Ward 提到“现在有一个可怕的 takeUntil 舞蹈，需要大量机器”，Shai Reznik 提到“Angular 处理一些订阅，如 http 和路由”。

作为回应，Ben 提到现在正在讨论允许 Observables 挂钩到 Angular 组件生命周期事件中，Ward 建议使用一个生命周期事件的 Observable，组件可以订阅该 Observable 作为了解何时完成作为组件内部状态维护的 Observables 的一种方式。

也就是说，我们现在主要需要解决方案，所以这里有一些其他资源。

RxJs 核心团队成员 Nicholas Jamieson 对 takeUntil() 模式的建议和一个 TSLint 规则来帮助执行它： https://ncjamieson.com/avoiding-takeuntil-leaks/ 轻量级 npm 包，它公开了一个 Observable 操作符，该操作符接受一个组件实例（this）作为参数并在 ngOnDestroy 期间自动取消订阅：https://github.com/NetanelBasal/ngx-take-until-destroy 如果您不进行 AOT 构建，则上述另一种变体具有更好的人体工程学（但我们应该现在都在做 AOT）：https://github.com/smnbbrv/ngx-rx-collector 自定义指令 *ngSubscribe 像异步管道一样工作，但在模板中创建一个嵌入式视图，因此您可以在整个过程中引用“未包装”值您的模板：https://netbasal.com/diy-subscription-handling-directive-in-angular-c8f6e762697f

我在 Nicholas 博客的评论中提到，过度使用 takeUntil() 可能表明您的组件正在尝试做太多事情，并且将您现有的组件分为 Feature 和 Presentational 组件应予以考虑。然后，您可以将特征组件中的 Observable | async 放入展示组件的 Input 中，这意味着在任何地方都不需要订阅。阅读有关此方法的更多信息here。

---

您无需拥有大量订阅并手动取消订阅。使用 Subject 和 takeUntil 组合来像老板一样处理订阅：

import { Subject } from "rxjs"
import { takeUntil } from "rxjs/operators"

@Component({
  moduleId: __moduleName,
  selector: "my-view",
  templateUrl: "../views/view-route.view.html"
})
export class ViewRouteComponent implements OnInit, OnDestroy {
  componentDestroyed$: Subject<boolean> = new Subject()

  constructor(private titleService: TitleService) {}

  ngOnInit() {
    this.titleService.emitter1$
      .pipe(takeUntil(this.componentDestroyed$))
      .subscribe((data: any) => { /* ... do something 1 */ })

    this.titleService.emitter2$
      .pipe(takeUntil(this.componentDestroyed$))
      .subscribe((data: any) => { /* ... do something 2 */ })

    //...

    this.titleService.emitterN$
      .pipe(takeUntil(this.componentDestroyed$))
      .subscribe((data: any) => { /* ... do something N */ })
  }

  ngOnDestroy() {
    this.componentDestroyed$.next(true)
    this.componentDestroyed$.complete()
  }
}

替代方法，在 by @acumartini in comments 中提出，使用 takeWhile 而不是 takeUntil。您可能更喜欢它，但请注意，这样您的 Observable 执行将不会在组件的 ngDestroy 上被取消（例如，当您进行耗时的计算或等待来自服务器的数据时）。基于takeUntil的方法没有这个缺点，会导致请求立即取消。 Thanks to @AlexChe for detailed explanation in comments。

所以这里是代码：

@Component({
  moduleId: __moduleName,
  selector: "my-view",
  templateUrl: "../views/view-route.view.html"
})
export class ViewRouteComponent implements OnInit, OnDestroy {
  alive: boolean = true

  constructor(private titleService: TitleService) {}

  ngOnInit() {
    this.titleService.emitter1$
      .pipe(takeWhile(() => this.alive))
      .subscribe((data: any) => { /* ... do something 1 */ })

    this.titleService.emitter2$
      .pipe(takeWhile(() => this.alive))
      .subscribe((data: any) => { /* ... do something 2 */ })

    // ...

    this.titleService.emitterN$
      .pipe(takeWhile(() => this.alive))
      .subscribe((data: any) => { /* ... do something N */ })
  }

  ngOnDestroy() {
    this.alive = false
  }
}

---

Subscription 类有一个有趣的特性：

表示一次性资源，例如 Observable 的执行。订阅有一个重要的方法，取消订阅，它不接受任何参数，只是释放订阅持有的资源。此外，订阅可以通过 add() 方法组合在一起，该方法会将子订阅附加到当前订阅。当订阅被取消订阅时，它的所有子（及其孙子）也将被取消订阅。

您可以创建一个聚合订阅对象，将您的所有订阅分组。为此，您可以创建一个空订阅并使用其 add() 方法向其添加订阅。当您的组件被销毁时，您只需要取消订阅聚合订阅。

@Component({ ... })
export class SmartComponent implements OnInit, OnDestroy {
  private subscriptions = new Subscription();

  constructor(private heroService: HeroService) {
  }

  ngOnInit() {
    this.subscriptions.add(this.heroService.getHeroes().subscribe(heroes => this.heroes = heroes));
    this.subscriptions.add(/* another subscription */);
    this.subscriptions.add(/* and another subscription */);
    this.subscriptions.add(/* and so on */);
  }

  ngOnDestroy() {
    this.subscriptions.unsubscribe();
  }
}

---

关于 Angular 组件内的 observables 取消订阅的一些最佳实践：

来自Routing & Navigation的报价

当订阅组件中的 observable 时，您几乎总是安排在组件被销毁时取消订阅。有一些特殊的 observables 是不必要的。 ActivatedRoute 可观察对象属于例外。 ActivatedRoute 及其 observables 与 Router 本身是绝缘的。当不再需要路由组件并且注入的 ActivatedRoute 随之消亡时，Router 会销毁路由组件。无论如何，请随时取消订阅。这是无害的，绝不是坏习惯。

并回应以下链接：

(1) 我应该取消订阅 Angular 2 Http Observables 吗？

(2) Http 方法创建的 observables 是否需要取消订阅？

(3) RxJS：不要退订

(4) Angular 中取消订阅 Observables 的最简单方法

(5) RxJS 退订文档

(6) 取消订阅服务是没有意义的，因为没有内存泄漏的机会

(7) 我们是否需要取消订阅完成/错误输出的可观察对象？

(8) 关于 http observable 的评论

我收集了一些关于 Angular 组件中的 observables 取消订阅的最佳实践与您分享：

http observable 取消订阅是有条件的，我们应该根据具体情况考虑在组件被销毁后运行的“订阅回调”的影响。我们知道 Angular 取消订阅并清理 http observable 本身 (1), (2)。虽然从资源的角度来看这是正确的，但它只说明了一半。假设我们正在讨论从组件内直接调用 http，并且 http 响应花费的时间比需要的要长，因此用户关闭了组件。即使组件被关闭和销毁，subscribe() 处理程序仍然会被调用。这可能会产生不必要的副作用，并且在更糟糕的情况下会使应用程序状态中断。如果回调中的代码试图调用刚刚被处理掉的东西，它也可能导致异常。然而，有时它们也是需要的。例如，假设您正在创建一个电子邮件客户端，并且在电子邮件完成发送时触发声音 - 即使组件已关闭，您仍然希望这种情况发生 (8)。

无需取消订阅完成或错误的 observables。但是，这样做并没有什么坏处(7)。

尽可能使用 AsyncPipe，因为它会在组件销毁时自动取消订阅 observable。

如果在嵌套（使用组件选择器在 tpl 中添加）或动态组件中订阅了诸如 route.params 之类的 ActivatedRoute 可观察对象，则取消订阅它们，因为只要父/主机组件存在，它们就可能被订阅多次。如上述路由和导航文档中的引用所述，无需在其他情况下取消订阅它们。

取消订阅通过 Angular 服务公开的组件之间共享的全局 observables，例如，只要组件初始化，它们就可能被多次订阅。

无需取消订阅应用程序范围服务的内部可观察对象，因为该服务永远不会被销毁，除非您的整个应用程序被销毁，否则没有真正的理由取消订阅，也不会出现内存泄漏。 (6)。注意：关于作用域服务，即组件提供者，它们在组件被销毁时被销毁。在这种情况下，如果我们订阅了这个提供者中的任何 observable，我们应该考虑使用 OnDestroy 生命周期钩子取消订阅，该钩子将在服务被销毁时调用，根据文档。

使用抽象技术来避免可能因取消订阅而导致的任何代码混乱。您可以使用 takeUntil (3) 管理您的订阅，也可以使用 (4) 取消订阅 Angular 中的 Observables 的最简单方法中提到的这个 npm 包。

始终取消订阅 FormGroup 可观察对象，例如 form.valueChanges 和 form.statusChanges

总是退订像 renderer2.listen 这样的 Renderer2 服务的 observables

取消订阅所有其他可观察对象作为内存泄漏保护步骤，直到 Angular 文档明确告诉我们哪些可观察对象不需要取消订阅（检查问题：（5）RxJS 取消订阅（打开）的文档）。

奖励：始终使用 Angular 方法来绑定诸如 HostListener 之类的事件，因为 Angular 非常关心在需要时删除事件侦听器并防止由于事件绑定而导致的任何潜在内存泄漏。

一个很好的最后提示：如果您不知道 observable 是否被自动取消订阅/完成，请将 complete 回调添加到 subscribe(...) 并检查它是否在组件被调用时被调用被摧毁。

---

这取决于。如果通过调用 someObservable.subscribe()，您开始占用一些必须在组件生命周期结束时手动释放的资源，那么您应该调用 theSubscription.unsubscribe() 以防止内存泄漏。

让我们仔细看看你的例子：

getHero() 返回 http.get() 的结果。如果您查看 angular 2 source code，http.get() 会创建两个事件侦听器：

_xhr.addEventListener('load', onLoad);
_xhr.addEventListener('error', onError);

通过调用 unsubscribe()，您可以取消请求以及侦听器：

_xhr.removeEventListener('load', onLoad);
_xhr.removeEventListener('error', onError);
_xhr.abort();

请注意，_xhr 是特定于平台的，但我认为在您的情况下假设它是 XMLHttpRequest() 是安全的。

通常，这足以证明手动调用 unsubscribe()。但是根据这个WHATWG spec，一旦“完成”，XMLHttpRequest() 就会受到垃圾回收的影响，即使它附加了事件侦听器。所以我想这就是为什么 Angular 2 官方指南省略了 unsubscribe() 并让 GC 清理侦听器的原因。

至于您的第二个示例，它取决于 params 的实现。到今天为止，Angular 官方指南不再显示退订 params。我再次查看 src，发现 params 只是一个 BehaviorSubject。由于没有使用事件侦听器或计时器，也没有创建全局变量，因此省略 unsubscribe() 应该是安全的。

您的问题的底线是始终调用 unsubscribe() 以防止内存泄漏，除非您确定 observable 的执行不会创建全局变量、添加事件侦听器、设置计时器或执行任何其他导致在内存泄漏中。

如有疑问，请查看该 observable 的实现。如果 observable 在其 unsubscribe() 中写入了一些清理逻辑，这通常是构造函数返回的函数，那么您有充分的理由认真考虑调用 unsubscribe()。

---

Angular 2 官方文档解释了何时取消订阅以及何时可以安全地忽略它。看看这个链接：

https://angular.io/docs/ts/latest/cookbook/component-communication.html#!#bidirectional-service

查找标题为 Parent and children 通过服务进行通信的段落，然后是蓝色框：

请注意，当 AstronautComponent 被销毁时，我们会捕获订阅并取消订阅。这是一个内存泄漏保护步骤。这个应用程序没有实际风险，因为 AstronautComponent 的生命周期与应用程序本身的生命周期相同。在更复杂的应用程序中，这并不总是正确的。我们没有将这个守卫添加到 MissionControlComponent 中，因为作为父级，它控制着 MissionService 的生命周期。

我希望这可以帮助你。

---

基于：Using Class inheritance to hook to Angular 2 component lifecycle

另一种通用方法：

导出抽象类 UnsubscribeOnDestroy 实现 OnDestroy { protected d$: Subject;构造函数() { this.d$ = new Subject();常量 f = this.ngOnDestroy; this.ngOnDestroy = () => { f(); this.d$.next(); this.d$.complete(); }; } public ngOnDestroy() { // 无操作 } }

并使用：

@Component({ selector: 'my-comp', template: `` }) 导出类 RsvpFormSaveComponent 扩展 UnsubscribeOnDestroy 实现 OnInit { constructor() { super(); } ngOnInit(): void { Observable.of('bla') .takeUntil(this.d$) .subscribe(val => console.log(val)); } }

---

由于 seangwright 的解决方案（编辑 3）看起来非常有用，我也发现将这个功能打包到基础组件中很痛苦，并提示其他项目团队记住在 ngOnDestroy 上调用 super() 来激活这个功能。

这个答案提供了一种摆脱超级调用的方法，并使“componentDestroyed$”成为基础组件的核心。

class BaseClass {
    protected componentDestroyed$: Subject<void> = new Subject<void>();
    constructor() {

        /// wrap the ngOnDestroy to be an Observable. and set free from calling super() on ngOnDestroy.
        let _$ = this.ngOnDestroy;
        this.ngOnDestroy = () => {
            this.componentDestroyed$.next();
            this.componentDestroyed$.complete();
            _$();
        }
    }

    /// placeholder of ngOnDestroy. no need to do super() call of extended class.
    ngOnDestroy() {}
}

然后您可以自由使用此功能，例如：

@Component({
    selector: 'my-thing',
    templateUrl: './my-thing.component.html'
})
export class MyThingComponent extends BaseClass implements OnInit, OnDestroy {
    constructor(
        private myThingService: MyThingService,
    ) { super(); }

    ngOnInit() {
        this.myThingService.getThings()
            .takeUntil(this.componentDestroyed$)
            .subscribe(things => console.log(things));
    }

    /// optional. not a requirement to implement OnDestroy
    ngOnDestroy() {
        console.log('everything works as intended with or without super call');
    }

}

---

官方编辑#3 答案（和变体）效果很好，但让我感到困惑的是围绕可观察订阅的业务逻辑的“混乱”。

这是使用包装器的另一种方法。

警告：实验代码

文件 subscribeAndGuard.ts 用于创建一个新的 Observable 扩展来包装 .subscribe() 并在其中包装 ngOnDestroy()。
用法与 .subscribe() 相同，除了额外的第一个引用组件的参数。

import { Observable } from 'rxjs/Observable';
import { Subscription } from 'rxjs/Subscription';

const subscribeAndGuard = function(component, fnData, fnError = null, fnComplete = null) {

  // Define the subscription
  const sub: Subscription = this.subscribe(fnData, fnError, fnComplete);

  // Wrap component's onDestroy
  if (!component.ngOnDestroy) {
    throw new Error('To use subscribeAndGuard, the component must implement ngOnDestroy');
  }
  const saved_OnDestroy = component.ngOnDestroy;
  component.ngOnDestroy = () => {
    console.log('subscribeAndGuard.onDestroy');
    sub.unsubscribe();
    // Note: need to put original back in place
    // otherwise 'this' is undefined in component.ngOnDestroy
    component.ngOnDestroy = saved_OnDestroy;
    component.ngOnDestroy();

  };

  return sub;
};

// Create an Observable extension
Observable.prototype.subscribeAndGuard = subscribeAndGuard;

// Ref: https://www.typescriptlang.org/docs/handbook/declaration-merging.html
declare module 'rxjs/Observable' {
  interface Observable<T> {
    subscribeAndGuard: typeof subscribeAndGuard;
  }
}

这是一个有两个订阅的组件，一个有包装，一个没有。唯一需要注意的是它必须实现 OnDestroy（如果需要，可以使用空主体），否则 Angular 不知道调用包装版本。

import { Component, OnInit, OnDestroy } from '@angular/core';
import { Observable } from 'rxjs/Observable';
import 'rxjs/Rx';
import './subscribeAndGuard';

@Component({
  selector: 'app-subscribing',
  template: '<h3>Subscribing component is active</h3>',
})
export class SubscribingComponent implements OnInit, OnDestroy {

  ngOnInit() {

    // This subscription will be terminated after onDestroy
    Observable.interval(1000)
      .subscribeAndGuard(this,
        (data) => { console.log('Guarded:', data); },
        (error) => { },
        (/*completed*/) => { }
      );

    // This subscription will continue after onDestroy
    Observable.interval(1000)
      .subscribe(
        (data) => { console.log('Unguarded:', data); },
        (error) => { },
        (/*completed*/) => { }
      );
  }

  ngOnDestroy() {
    console.log('SubscribingComponent.OnDestroy');
  }
}

一个演示 plunker 是 here

附加说明：重新编辑 3 - “官方”解决方案，这可以通过在订阅前使用 takeWhile() 而不是 takeUntil() 以及在 ngOnDestroy 中使用简单的布尔值而不是另一个 Observable 来简化。

@Component({...})
export class SubscribingComponent implements OnInit, OnDestroy {

  iAmAlive = true;
  ngOnInit() {

    Observable.interval(1000)
      .takeWhile(() => { return this.iAmAlive; })
      .subscribe((data) => { console.log(data); });
  }

  ngOnDestroy() {
    this.iAmAlive = false;
  }
}

---

对于在发出 AsyncSubject 之类的结果后直接完成的可观察对象，或者例如来自 http 请求的可观察对象，您不需要取消订阅。为这些调用 unsubscribe() 并没有什么坏处，但如果 observable 是 closed，则取消订阅方法 will simply not do anything：

if (this.closed) {
  return;
}

当您拥有随时间发出多个值的长期可观察对象（例如 BehaviorSubject 或 ReplaySubject）时，您需要取消订阅以防止内存泄漏。

您可以轻松地创建一个可观察对象，该可观察对象在使用管道运算符从此类长期存在的可观察对象发出结果后直接完成。在这里的一些答案中提到了 take(1) 管道。但我更喜欢the first() pipe。与 take(1) 的区别在于它将：

如果 Observable 在发送任何下一个通知之前完成，则将 EmptyError 传递给 Observer 的错误回调。

第一个管道的另一个优点是您可以传递一个谓词，该谓词将帮助您返回满足某些条件的第一个值：

const predicate = (result: any) => { 
  // check value and return true if it is the result that satisfies your needs
  return true;
}
observable.pipe(first(predicate)).subscribe(observer);

First 将在发出第一个值后直接完成（或在传递函数参数时满足您的谓词的第一个值），因此无需取消订阅。

有时你不确定你是否有一个长寿的 observable。我并不是说这是一种好的做法，但您可以随时添加 first 管道，以确保您不需要手动取消订阅。在只发出一个值的 observable 上添加一个额外的 first 管道并没有什么坏处。

在开发过程中，您可以使用 the single pipe，如果源 observable 发出多个事件，它将失败。这可以帮助您探索 observable 的类型以及是否有必要取消订阅它。

observable.pipe(single()).subscribe(observer);

first 和 single 看起来非常相似，两个管道都可以采用可选谓词，但差异很重要，并且在 this stackoverflow answer here 中得到了很好的总结：

First 将在第一个项目出现时立即发出。之后会马上完成。如果源 observable 发出多个事件，则 Single 将失败。

请注意，我尝试在参考官方文档的情况下尽可能准确和完整地回答，但如果缺少重要内容，请发表评论......

---

根据 @seangwright 的回答，我编写了一个抽象类来处理组件中的“无限”可观察对象的订阅：

import { OnDestroy } from '@angular/core';
import { Subscription } from 'rxjs/Subscription';
import { Subject } from 'rxjs/Subject';
import { Observable } from 'rxjs/Observable';
import { PartialObserver } from 'rxjs/Observer';

export abstract class InfiniteSubscriberComponent implements OnDestroy {
  private onDestroySource: Subject<any> = new Subject();

  constructor() {}

  subscribe(observable: Observable<any>): Subscription;

  subscribe(
    observable: Observable<any>,
    observer: PartialObserver<any>
  ): Subscription;

  subscribe(
    observable: Observable<any>,
    next?: (value: any) => void,
    error?: (error: any) => void,
    complete?: () => void
  ): Subscription;

  subscribe(observable: Observable<any>, ...subscribeArgs): Subscription {
    return observable
      .takeUntil(this.onDestroySource)
      .subscribe(...subscribeArgs);
  }

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

要使用它，只需在您的 Angular 组件中扩展它并调用 subscribe() 方法，如下所示：

this.subscribe(someObservable, data => doSomething());

它还像往常一样接受错误和完成回调，观察者对象，或者根本不接受回调。如果您也在子组件中实现该方法，请记住调用 super.ngOnDestroy()。

在此处查找 Ben Lesh 的其他参考资料：RxJS: Don’t Unsubscribe。

---

Subscription 本质上只有一个 unsubscribe() 函数来释放资源或取消 Observable 执行。在 Angular 中，当组件被销毁时，我们必须取消订阅 Observable。幸运的是，Angular 有一个 ngOnDestroy 钩子，它在组件被销毁之前被调用，这使开发人员能够在这里提供清理人员以避免挂起订阅、打开门户以及将来可能会在后面咬我们的东西

@Component({...})
export class AppComponent implements OnInit, OnDestroy {
    subscription: Subscription 
    ngOnInit () {
        var observable = Rx.Observable.interval(1000);
        this.subscription = observable.subscribe(x => console.log(x));
    }
    ngOnDestroy() {
        this.subscription.unsubscribe()
    }
}

我们将 ngOnDestroy 添加到 AppCompoennt 并在 this.subscription Observable 上调用 unsubscribe 方法

如果有多个订阅：

@Component({...})
export class AppComponent implements OnInit, OnDestroy {
    subscription1$: Subscription
    subscription2$: Subscription 
    ngOnInit () {
        var observable1$ = Rx.Observable.interval(1000);
        var observable2$ = Rx.Observable.interval(400);
        this.subscription1$ = observable.subscribe(x => console.log("From interval 1000" x));
        this.subscription2$ = observable.subscribe(x => console.log("From interval 400" x));
    }
    ngOnDestroy() {
        this.subscription1$.unsubscribe()
        this.subscription2$.unsubscribe()
    }
}

---

我尝试了 seangwright 的解决方案（编辑 3）

这不适用于由计时器或间隔创建的 Observable。

但是，我通过使用另一种方法让它工作：

import { Component, OnDestroy, OnInit } from '@angular/core';
import 'rxjs/add/operator/takeUntil';
import { Subject } from 'rxjs/Subject';
import { Subscription } from 'rxjs/Subscription';
import 'rxjs/Rx';

import { MyThingService } from '../my-thing.service';

@Component({
   selector: 'my-thing',
   templateUrl: './my-thing.component.html'
})
export class MyThingComponent implements OnDestroy, OnInit {
   private subscriptions: Array<Subscription> = [];

  constructor(
     private myThingService: MyThingService,
   ) { }

  ngOnInit() {
    const newSubs = this.myThingService.getThings()
        .subscribe(things => console.log(things));
    this.subscriptions.push(newSubs);
  }

  ngOnDestroy() {
    for (const subs of this.subscriptions) {
      subs.unsubscribe();
   }
 }
}

---

我喜欢最后两个答案，但如果子类在 ngOnDestroy 中引用 "this"，我会遇到问题。

我把它改成了这个，看起来它解决了这个问题。

export abstract class BaseComponent implements OnDestroy {
    protected componentDestroyed$: Subject<boolean>;
    constructor() {
        this.componentDestroyed$ = new Subject<boolean>();
        let f = this.ngOnDestroy;
        this.ngOnDestroy = function()  {
            // without this I was getting an error if the subclass had
            // this.blah() in ngOnDestroy
            f.bind(this)();
            this.componentDestroyed$.next(true);
            this.componentDestroyed$.complete();
        };
    }
    /// placeholder of ngOnDestroy. no need to do super() call of extended class.
    ngOnDestroy() {}
}

---

当组件被销毁时，您通常需要取消订阅，但随着我们的进行，Angular 会越来越多地处理它，例如在 Angular4 的新次要版本中，他们有这个部分用于路由取消订阅：

您需要退订吗？如 ActivatedRoute：Routing & Navigation 页面的路线信息的一站式商店部分所述，Router 管理它提供的 observables 并本地化订阅。当组件被销毁时，订阅会被清理，以防止内存泄漏，因此您不需要从路由 paramMap Observable 取消订阅。

下面的例子也是 Angular 创建组件并在之后销毁它的一个很好的例子，看看组件是如何实现 OnDestroy 的，如果你需要 onInit，你也可以在你的组件中实现它，比如 implements OnInit, OnDestroy

import { Component, Input, OnDestroy } from '@angular/core';  
import { MissionService } from './mission.service';
import { Subscription }   from 'rxjs/Subscription';

@Component({
  selector: 'my-astronaut',
  template: `
    <p>
      {{astronaut}}: <strong>{{mission}}</strong>
      <button
        (click)="confirm()"
        [disabled]="!announced || confirmed">
        Confirm
      </button>
    </p>
  `
})

export class AstronautComponent implements OnDestroy {
  @Input() astronaut: string;
  mission = '<no mission announced>';
  confirmed = false;
  announced = false;
  subscription: Subscription;

  constructor(private missionService: MissionService) {
    this.subscription = missionService.missionAnnounced$.subscribe(
      mission => {
        this.mission = mission;
        this.announced = true;
        this.confirmed = false;
    });
  }

  confirm() {
    this.confirmed = true;
    this.missionService.confirmMission(this.astronaut);
  }

  ngOnDestroy() {
    // prevent memory leak when component destroyed
    this.subscription.unsubscribe();
  }
}

---

上述情况的另一个简短补充是：

始终取消订阅，当订阅流中的新值不再需要或无关紧要时，它会导致触发器数量减少并在少数情况下提高性能。订阅的数据/事件不再存在或需要对全新流的新订阅（刷新等）的组件是取消订阅的一个很好的例子。

---

如果需要取消订阅，可以使用以下可观察管道方法的运算符

import { Observable, Subject } from 'rxjs';
import { takeUntil } from 'rxjs/operators';
import { OnDestroy } from '@angular/core';

export const takeUntilDestroyed = (componentInstance: OnDestroy) => <T>(observable: Observable<T>) => {
  const subjectPropertyName = '__takeUntilDestroySubject__';
  const originalOnDestroy = componentInstance.ngOnDestroy;
  const componentSubject = componentInstance[subjectPropertyName] as Subject<any> || new Subject();

  componentInstance.ngOnDestroy = (...args) => {
    originalOnDestroy.apply(componentInstance, args);
    componentSubject.next(true);
    componentSubject.complete();
  };

  return observable.pipe(takeUntil<T>(componentSubject));
};

它可以这样使用：

import { Component, OnDestroy, OnInit } from '@angular/core';
import { Observable } from 'rxjs';

@Component({ template: '<div></div>' })
export class SomeComponent implements OnInit, OnDestroy {

  ngOnInit(): void {
    const observable = Observable.create(observer => {
      observer.next('Hello');
    });

    observable
      .pipe(takeUntilDestroyed(this))
      .subscribe(val => console.log(val));
  }

  ngOnDestroy(): void {
  }
}

操作符封装了组件的 ngOnDestroy 方法。

重要提示：操作员应该是可观察管道中的最后一个操作员。

---

在 ngOnDestroy 函数 (angular lifeCycle) 的 SPA 应用程序中，对于每个订阅，您都需要取消订阅。优势 => 以防止状态变得太重。

例如：在组件 1 中：

import {UserService} from './user.service';

private user = {name: 'test', id: 1}

constructor(public userService: UserService) {
    this.userService.onUserChange.next(this.user);
}

在役：

import {BehaviorSubject} from 'rxjs/BehaviorSubject';

public onUserChange: BehaviorSubject<any> = new BehaviorSubject({});

在组件 2 中：

import {Subscription} from 'rxjs/Subscription';
import {UserService} from './user.service';

private onUserChange: Subscription;

constructor(public userService: UserService) {
    this.onUserChange = this.userService.onUserChange.subscribe(user => {
        console.log(user);
    });
}

public ngOnDestroy(): void {
    // note: Here you have to be sure to unsubscribe to the subscribe item!
    this.onUserChange.unsubscribe();
}

---

为了处理订阅，我使用“Unsubscriber”类。

这是取消订阅者类。

export class Unsubscriber implements OnDestroy {
  private subscriptions: Subscription[] = [];

  addSubscription(subscription: Subscription | Subscription[]) {
    if (Array.isArray(subscription)) {
      this.subscriptions.push(...subscription);
    } else {
      this.subscriptions.push(subscription);
    }
  }

  unsubscribe() {
    this.subscriptions
      .filter(subscription => subscription)
      .forEach(subscription => {
        subscription.unsubscribe();
      });
  }

  ngOnDestroy() {
    this.unsubscribe();
  }
}

您可以在任何组件/服务/效果等中使用此类。

例子：

class SampleComponent extends Unsubscriber {
    constructor () {
        super();
    }

    this.addSubscription(subscription);
}

---

SubSink 包，一个简单且一致的退订解决方案

由于没有其他人提到它，我想推荐由 Ward Bell 创建的 Subsink 包：https://github.com/wardbell/subsink#readme。

我一直在一个项目中使用它，我们是几个开发人员都在使用它。有一种适用于各种情况的一致方式非常有帮助。

---

出于性能原因，始终建议从您的可观察订阅中取消订阅以避免内存泄漏，并且有不同的方法可以做到这一点，

顺便说一句，我阅读了大部分答案，但我没有找到有人在谈论 async 管道，建议使用 Angular 应用程序的 Rxjs 模式，因为它提供自动订阅和订阅离开将被销毁的组件时：

请找到一个如何实施的例子

app.compoent.ts：

import { Component, OnInit } from '@angular/core';
import { Observable } from 'rxjs';

import { BookService } from './book.service';
import { Book } from './book';

@Component({
   selector: 'app-observable',
   templateUrl: './observable.component.html'
})
export class AppComponent implements OnInit { 
   books$: Observable<Book[]>
   constructor(private bookService: BookService) { }
   ngOnInit(): void {
        this.books$ = this.bookService.getBooksWithObservable();
   }
} 

app.compoent.html：

<h3>AsyncPipe with Promise using NgFor</h3>
<ul>
  <li *ngFor="let book of books$ | async" >
    Id: {{book?.id}}, Name: {{book?.name}}
  </li>
</ul>

---

您可以使用最新的 Subscription 类来取消订阅 Observable，而代码不会那么混乱。

我们可以使用 normal variable 执行此操作，但在每次新订阅时都会为 override the last subscription，因此请避免这种情况，当您处理更多数量的 Obseravable 和诸如 {1 之类的 Observable 类型时，这种方法非常有用} 和 Subject

订阅

表示一次性资源，例如 Observable 的执行。订阅有一个重要的方法，取消订阅，它不接受任何参数，只是释放订阅持有的资源。

你可以通过两种方式使用它，

可以直接将订阅推送到 Subscription Array subscriptions:Subscription[] = []; ngOnInit(): void { this.subscription.push(this.dataService.getMessageTracker().subscribe((param: any) => { //... })); this.subscription.push(this.dataService.getFileTracker().subscribe((param: any) => { //... })); } ngOnDestroy(){ // 防止组件销毁时内存泄漏 this.subscriptions.forEach(s => s.unsubscribe()); }

使用订阅的 add() 订阅 = new Subscription(); this.subscriptions.add(subscribeOne); this.subscriptions.add(subscribeTwo); ngOnDestroy() { this.subscriptions.unsubscribe(); }

Subscription 可以保留子订阅并安全地取消订阅它们。此方法处理可能的错误（例如，如果任何子订阅为空）。

希望这可以帮助.. ：）

---

就我而言，我正在使用@seanwright 提出的解决方案的变体：
https://github.com/NetanelBasal/ngx-take-until-destroy

它是 ngx-rocket / starter-kit 项目中使用的文件。您可以在此处访问它until-destroyed.ts

该组件看起来像这样

/**
 * RxJS operator that unsubscribe from observables on destory.
 * Code forked from https://github.com/NetanelBasal/ngx-take-until-destroy
 *
 * IMPORTANT: Add the `untilDestroyed` operator as the last one to
 * prevent leaks with intermediate observables in the
 * operator chain.
 *
 * @param instance The parent Angular component or object instance.
 * @param destroyMethodName The method to hook on (default: 'ngOnDestroy').
 */
import { untilDestroyed } from '../../core/until-destroyed';

@Component({
  selector: 'app-example',
  templateUrl: './example.component.html'
})
export class ExampleComponent implements OnInit, OnDestroy {

  ngOnInit() {
    interval(1000)
        .pipe(untilDestroyed(this))
        .subscribe(val => console.log(val));

    // ...
  }


  // This method must be present, even if empty.
  ngOnDestroy() {
    // To protect you, an error will be thrown if it doesn't exist.
  }
}

---

这里有很多很棒的答案...

让我添加另一个替代方案：

import { interval    } from "rxjs";
import { takeUntil   } from "rxjs/operators";
import { Component   } from "@angular/core";
import { Destroyable } from "@bespunky/angular-zen/core";

@Component({
    selector: 'app-no-leak-demo',
    template: '👍 Destroyable component rendered. Unload me and watch me cleanup...'
})
export class NoLeakComponent extends Destroyable
{
    constructor()
    {
        super();

        this.subscribeToInterval();
    }

    private subscribeToInterval(): void
    {
        const value    = interval(1000);
        const observer = {
            next    : value => console.log(`👍 Destroyable: ${value}`),
            complete: ()    => console.log('👍 Observable completed.')
        };

        // ==== Comment one and uncomment the other to see the difference ====
        
        // Subscribe using the inherited subscribe method
         this.subscribe(value, observer);

        // ... or pipe-in the inherited destroyed subject
        //value.pipe(takeUntil(this.destroyed)).subscribe(observer);
    }
}

Live Example

这里发生了什么事

组件/服务扩展了 Destroyable（来自名为 @bespunky/angular-zen 的库）。

该类现在可以简单地使用 this.subscribe() 或 takeUntil(this.destroyed) 而无需任何额外的样板代码。

要安装库，请使用： > npm install @bespunky/angular-zen

---

这是我对这个问题的看法，为了让我的生活保持简单，我选择了在组件被销毁时取消订阅的手动方式。

为此，我创建了一个名为 Subscriptor 的类，它主要包含静态成员，即：

一个私有变量订阅 - 它包含所有提供的订阅

订阅设置器 - 将每个新订阅推送到订阅数组

取消订阅方法 - 如果已定义，则取消订阅订阅数组包含的每个订阅，并清空订阅数组

下标.ts

import { Subscription } from "rxjs";

export class Subscriptor {
    private static subscriptions: Subscription[] = [];

    static set subscription(subscription: Subscription) {
        Subscriptor.subscriptions.push(subscription);
    }

    static unsubscribe() {
        Subscriptor.subscriptions.forEach(subscription => subscription ? subscription.unsubscribe() : 0);
        Subscriptor.subscriptions = [];
    }
}

组件内部的用法如下：

当您想订阅任何服务时，只需将订阅放到 Subscriptor 的 setter 中即可。

ngOnInit(): void {
    Subscriptor.subscription = this.userService.getAll().subscribe(users => this.users = users);
    Subscriptor.subscription = this.categoryService.getAll().subscribe(categories => this.categories = categories);
    Subscriptor.subscription = this.postService.getAll().subscribe(posts => this.posts = posts);
}

当您想取消订阅任何服务时，只需调用 Subscriptor 的取消订阅方法即可。

ngOnDestroy(): void {
    Subscriptor.unsubscribe();
}

---

处理袋

这个想法的灵感来自 RxSwift 的 DisposeBag，所以我决定开发一个类似但简单的结构。

DisposeBag 是一个数据结构，它包含对所有打开订阅的引用。它有助于在我们的组件中处理订阅，同时为我们提供 API 来跟踪打开订阅的状态。

优点

非常简单的 API，使您的代码看起来简单而小巧。提供用于跟踪开放订阅状态的 API（允许您显示不确定的进度条） 无依赖注入/包。

用法

在组件中：

@Component({
  selector: 'some-component',
  templateUrl: './some-component.component.html',
  changeDetection: ChangeDetectionStrategy.OnPush
})
export class SomeComponent implements OnInit, OnDestroy {

  public bag = new DisposeBag()
  
  constructor(private _change: ChangeDetectorRef) {
  }

  ngOnInit(): void {

    // an observable that takes some time to finish such as an api call.
    const aSimpleObservable = of(0).pipe(delay(5000))

    // this identifier allows us to track the progress for this specific subscription (very useful in template)
    this.bag.subscribe("submission", aSimpleObservable, () => { 
      this._change.markForCheck() // trigger UI change
     })
  }

  ngOnDestroy(): void {
    // never forget to add this line.
    this.bag.destroy()
  }
}

在模板中：

<!-- will be shown as long as the submission subscription is open -->
<span *ngIf="bag.inProgress('submission')">submission in progress</span>

<!-- will be shown as long as there's an open subscription in the bag  -->
<span *ngIf="bag.hasInProgress">some subscriptions are still in progress</span>

执行

import { Observable, Observer, Subject, Subscription, takeUntil } from "rxjs";


/**
 * This class facilitates the disposal of the subscription in our components.
 * instead of creating _unsubscribeAll and lots of boilerplates to create different variables for Subscriptions; 
 * you can just easily use subscribe(someStringIdentifier, observable, observer). then you can use bag.inProgress() with
 * the same someStringIdentifier on you html or elsewhere to determine the state of the ongoing subscription.
 *
 *  don't forget to add onDestroy() { this.bag.destroy() }
 * 
 *  Author: Hamidreza Vakilian (hvakilian1@gmail.com)
 * @export
 * @class DisposeBag
 */
export class DisposeBag {
    private _unsubscribeAll: Subject<any> = new Subject<any>();

    private subscriptions = new Map<string, Subscription>()


    /**
     * this method automatically adds takeUntil to your observable, adds it to a private map.
     * this method enables inProgress to work. don't forget to add onDestroy() { this.bag.destroy() }
     *
     * @template T
     * @param {string} id
     * @param {Observable<T>} obs
     * @param {Partial<Observer<T>>} observer
     * @return {*}  {Subscription}
     * @memberof DisposeBag
     */
    public subscribe<T>(id: string, obs: Observable<T>, observer: Partial<Observer<T>> | ((value: T) => void)): Subscription {
        if (id.isEmpty()) {
            throw new Error('disposable.subscribe is called with invalid id')
        }
        if (!obs) {
            throw new Error('disposable.subscribe is called with an invalid observable')
        }

        /* handle the observer */
        let subs: Subscription
        if (typeof observer === 'function') {
            subs = obs.pipe(takeUntil(this._unsubscribeAll)).subscribe(observer)
        } else if (typeof observer === 'object') {
            subs = obs.pipe(takeUntil(this._unsubscribeAll)).subscribe(observer)
        } else {
            throw new Error('disposable.subscribe is called with an invalid observer')
        }

        /* unsubscribe from the last possible subscription if in progress. */
        let possibleSubs = this.subscriptions.get(id)
        if (possibleSubs && !possibleSubs.closed) {
            console.info(`Disposebag: a subscription with id=${id} was disposed and replaced.`)
            possibleSubs.unsubscribe()
        }

        /* store the reference in the map */
        this.subscriptions.set(id, subs)

        return subs
    }


    /**
     * Returns true if any of the registered subscriptions is in progress.
     *
     * @readonly
     * @type {boolean}
     * @memberof DisposeBag
     */
    public get hasInProgress(): boolean {
        return Array.from(this.subscriptions.values()).reduce(
            (prev, current: Subscription) => { 
                return prev || !current.closed }
            , false)
    }

    /**
     * call this from your template or elsewhere to determine the state of each subscription.
     *
     * @param {string} id
     * @return {*} 
     * @memberof DisposeBag
     */
    public inProgress(id: string) {
        let possibleSubs = this.subscriptions.get(id)
        if (possibleSubs) {
            return !possibleSubs.closed
        } else {
            return false
        }
    }


    /**
     * Never forget to call this method in your onDestroy() method of your components.
     *
     * @memberof DisposeBag
     */
    public destroy() {
        this._unsubscribeAll.next(null);
        this._unsubscribeAll.complete();
    }
}

---

--- 更新 Angular 9 和 Rxjs 6 解决方案

在 Angular 组件的 ngDestroy 生命周期中使用取消订阅

class SampleComponent implements OnInit, OnDestroy {
  private subscriptions: Subscription;
  private sampleObservable$: Observable<any>;

  constructor () {}

  ngOnInit(){
    this.subscriptions = this.sampleObservable$.subscribe( ... );
  }

  ngOnDestroy() {
    this.subscriptions.unsubscribe();
  }
}

在 Rxjs 中使用 takeUntil

class SampleComponent implements OnInit, OnDestroy {
  private unsubscribe$: new Subject<void>;
  private sampleObservable$: Observable<any>;

  constructor () {}

  ngOnInit(){
    this.subscriptions = this.sampleObservable$
    .pipe(takeUntil(this.unsubscribe$))
    .subscribe( ... );
  }

  ngOnDestroy() {
    this.unsubscribe$.next();
    this.unsubscribe$.complete();
  }
}

对于您在 ngOnInit 调用的某些操作，该操作仅在组件初始化时发生一次。

class SampleComponent implements OnInit {

  private sampleObservable$: Observable<any>;

  constructor () {}

  ngOnInit(){
    this.subscriptions = this.sampleObservable$
    .pipe(take(1))
    .subscribe( ... );
  }
}

我们还有 async 管道。但是，这个在模板上使用（不在 Angular 组件中）。

---