在我的情况下，我使用了@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.
  }
}

出于性能原因，总是建议从你的可观察订阅中取消订阅，以避免内存泄漏，有不同的方法，

顺便说一下，我读了大部分的答案，我没有发现有人在谈论异步管道，推荐在Angular应用中使用Rxjs模式，因为它在离开组件时会自动提供订阅和订阅，而这些订阅将被销毁:

请找到一个如何实现它的例子

app.compoennt.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.compoennt.html:

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

下面是我对这个问题的看法，保持我的生活简单，我选择手动方式取消订阅时，组件被破坏。

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

私有变量subscriptions——它保存所有提供的订阅 订阅设置器——将每个新订阅推送到订阅数组 一个取消订阅方法——如果定义了订阅数组，则取消订阅数组中包含的每个订阅，并清空订阅数组

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的unsubscribe方法。

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

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

总是取消订阅，当订阅流中的新值不再需要或无关紧要时，它将导致更少的触发器数量，并在某些情况下提高性能。订阅的数据/事件不再存在，或者需要对全新流进行新订阅(刷新等)的组件就是取消订阅的好例子。

你不需要有一堆订阅和取消手动订阅。使用Subject和takeUntil组合来像boss一样处理订阅:

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

@acumartini在评论中提出了另一种方法，使用takeWhile而不是takeUntil。你可能更喜欢它，但要注意，这样你的Observable的执行就不会在组件的ngDestroy上被取消(例如，当你进行耗时的计算或等待来自服务器的数据时)。方法没有这个缺陷，它会导致立即取消请求。感谢@AlexChe在评论中的详细解释。

代码如下:

@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
  }
}

DisposeBag

这个想法的灵感来自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();
    }
}