typescript - 在 Angular 中使用 RxJs 时，我是否只需要完成组件中创建的 Observables？

基本上，我仍然不太确定在 Angular 组件中如何清理响应式代码。以下规则是否正确且足够？

我对于响应式组件的规则：

• 关于手动 x.subscribe()：

  • 如果使用 x.takeUntil(...) 或类似的结构，则无需手动清理
  • 如果在组件作用域中调用 x.complete()，则无需手动清理
  • 否则根据需要或在 ngOnDestroy() 中手动 .unsubscribe()

• 在模板的 observable | async 上：

  • 无需手动清理，并且兼容 onPush

• 关于像 new Subject() 这样手动直接创建 Observable:

  • 在 ngOnDestroy() 中手动 .complete()，否则永远不会关闭

• 关于使用其他 Observable 手动创建，比如 Observable.merge(...):

  • 只要订阅被清理，就不需要手动清理

我不太确定最后一点。

这是我根据这些规则重构的一个完整的真实组件示例，有人能看看是否存在内存泄漏的风险吗？所有公共的 $observables 都通过模板中的 async 管道订阅，所以我确信这些订阅已经被处理了。使用 new 创建的 subjects 会手动清理，所以应该没问题。我担心的部分是 Observable.combineLatest(...) 观察对象。

import { ChangeDetectionStrategy, Component, Input, OnDestroy } from '@angular/core';
import { Router } from '@angular/router';
import { DeliveryPointId } from 'app/bso';
import { BookmarkService } from 'app/general';
import { EAccess, EPermission, ReduxGetters, ReduxService } from 'app/redux';
import * as routing from 'app/routing';
import { BehaviorSubject, Observable } from 'app/rx';

@Component({
  selector: 'app-last-opened-workitems-widget',
  templateUrl: './last-opened-workitems-widget.component.html',
  changeDetection: ChangeDetectionStrategy.OnPush,
})
export class LastOpenedWorkitemsWidgetComponent implements OnDestroy {

  private readonly showPartners$ = new BehaviorSubject(true);
  private readonly showServices$ = new BehaviorSubject(true);
  private readonly showTens$ = new BehaviorSubject(true);
  private readonly showWorklist$ = new BehaviorSubject(true);

  @Input() set showItems(val: boolean) { this.showWorklist$.next(!!val); }
  @Input() set showPartners(val: boolean) { this.showPartners$.next(!!val); }
  @Input() set showProcesses(val: boolean) { this.showServices$.next(!!val); }
  @Input() set showTens(val: boolean) { this.showTens$.next(!!val); }

  constructor(
    private readonly redux: ReduxService,
    private readonly bookmarks: BookmarkService,
    private readonly router: Router,
  ) { }

  canPartners$ = this.redux.watch(ReduxGetters.userAccess).map(access => access[EPermission.Partner] >= EAccess.Read);
  canServices$ = this.redux.watch(ReduxGetters.userAccess).map(access => access[EPermission.Services] >= EAccess.Read);
  canTens$ = this.redux.watch(ReduxGetters.userAccess).map(access => access[EPermission.Tens] >= EAccess.Read);
  canWorklist$ = this.redux.watch(ReduxGetters.userAccess).map(access => access[EPermission.Worklist] >= EAccess.Read);

  lastItems$ = this.bookmarks.onLastOpenedWorkitems.map(ii => [...ii].reverse());
  lastPartners$ = this.bookmarks.onLastOpenedPartners.map(ii => [...ii].reverse());
  lastProcesses$ = this.bookmarks.onLastOpenedProcesses.map(ii => [...ii].reverse());
  lastTens$ = this.bookmarks.onLastOpenedTens.map(ii => [...ii].reverse());

  hasContentPartners$ = Observable
    .combineLatest(
      this.showPartners$.distinctUntilChanged(),
      this.canPartners$,
      this.lastPartners$.map(ii => ii.length > 0))
    .map(oks => oks.every(ii => ii));

  hasContentServices$ = Observable
    .combineLatest(
      this.showServices$.distinctUntilChanged(),
      this.canServices$,
      this.lastProcesses$.map(ii => ii.length > 0))
    .map(oks => oks.every(ii => ii));

  hasContentTens$ = Observable
    .combineLatest(
      this.showTens$.distinctUntilChanged(),
      this.canTens$,
      this.lastTens$.map(ii => ii.length > 0))
    .map(oks => oks.every(ii => ii));

  hasContentWorklist$ = Observable
    .combineLatest(
      this.showWorklist$.distinctUntilChanged(),
      this.canWorklist$,
      this.lastItems$.map(ii => ii.length > 0))
    .map(oks => oks.every(ii => ii));

  hasContent$ = Observable
    .combineLatest(this.hasContentPartners$, this.hasContentServices$, this.hasContentTens$, this.hasContentWorklist$)
    .map(oks => oks.some(ii => ii));

  ngOnDestroy() {
    [this.showPartners$, this.showServices$, this.showTens$, this.showWorklist$].forEach(ii => ii.complete());
  }

  gotoPartner = (id: string) => routing.gotoPartnerItem(this.router, id);
  gotoProcess = (id: number) => routing.gotoProcess(this.router, id);
  gotoTensItem = (id: DeliveryPointId) => routing.gotoTensItem(this.router, id);
  gotoWorkitem = (id: number) => routing.gotoWorkitem(this.router, id);

}

答案＃1

你需要取消你每一个订阅，例如在 ngOnDestroy() 方法中。

或者你可以使用诸如 takeUntil()、take()、first() 等操作符来“限制”可观察对象。

上述规则同样适用于 combineLatest。

不用担心由 AsyncPipe 创建的订阅，因为 AsyncPipe 本身负责取消自己订阅的订阅。

关于这个主题的一个不错的阅读是由 RxJS 负责人 Ben Lesh 撰写的文章。

答案＃2

好吧，我实际上刚用 .finally(...) 测试了一下，同时为所有中间 Observable 添加了日志，当组件被销毁时，它确实会被正确触发。

因此，我声明上述规则是正确且充分的。