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

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

重点:操作符应该在可观察管道的最后一个。

根据@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。

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

顺便说一下，我读了大部分的答案，我没有发现有人在谈论异步管道，推荐在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>

你不需要有一堆订阅和取消手动订阅。使用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
  }
}

更新Angular 9和Rxjs 6解决方案

在Angular组件的ngDestroy生命周期中使用unsubscribe

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调用的一些动作，在组件init时只会发生一次。

class SampleComponent implements OnInit {

  private sampleObservable$: Observable<any>;

  constructor () {}

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

我们也有async管道。但是，这个是在模板上使用的(不是在Angular组件中)。

你可以使用最新的订阅类来取消对Observable的订阅，代码不会那么乱。

我们可以用普通变量来做这个，但它会在每次新的订阅上覆盖上一次订阅，所以要避免这种情况，当你处理更多数量的可观察对象时，这种方法是非常有用的，以及像behaviour Subject和Subject这样的可观察对象类型

订阅

表示一个一次性资源，比如Observable的执行。订阅有一个重要的方法，即unsubscribe，它不接受参数，只处理订阅所持有的资源。

你可以用两种方式来使用它，

you can directly push the subscription to 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(){ // prevent memory leak when component destroyed this.subscriptions.forEach(s => s.unsubscribe()); } using add() of Subscription subscriptions = new Subscription(); this.subscriptions.add(subscribeOne); this.subscriptions.add(subscribeTwo); ngOnDestroy() { this.subscriptions.unsubscribe(); }

订阅可以保存子订阅并安全地取消所有订阅。这个方法处理可能的错误(例如，如果任何子订阅为空)。

希望这能有所帮助。：）