Angular11使用@ngneat/until-destroy报ObjectUnsubscribedError如何解决

错误含义 #

ObjectUnsubscribedError是RxJS内置的标准错误，当代码尝试对已经执行过unsubscribe()、处于关闭状态的Subject/可观察对象调用next()、error()、complete()方法时，就会抛出该错误。

触发原因 #

你项目中开启了@UntilDestroy({ checkProperties: true })配置，该配置会在组件销毁时，自动扫描组件实例上所有属于Observable/Subject/Subscription类型的属性，统一执行unsubscribe()操作。
你在组件中声明的widget$、itemHistory$、values$三个ReplaySubject都是组件的公开属性，会被until-destroy自动取消订阅、标记为关闭状态。而组件内的异步逻辑（比如带debounceTime(500)的流、store的异步查询）存在延时，当组件销毁后这些异步回调才触发时，就会尝试调用已经关闭的Subject的next()方法，进而抛出大量报错。

规避方案 #

有三种常用方案可以解决该问题，可根据项目情况选择：

方案1：跳过指定Subject的自动销毁逻辑 #

@ngneat/until-destroy提供了@skipCheck装饰器，用来标记不需要被自动取消订阅的Subject属性，避免组件销毁时Subject被误关闭。
代码示例：

import { UntilDestroy, skipCheck } from '@ngneat/until-destroy';

@UntilDestroy({ checkProperties: true })
@Component({
  selector: 'diagram-card',
  template: `<widget-diagram [data$]="values$"></widget-diagram>`,
  styleUrls: ['./diagram-card.component.scss'],
})
export class DiagramCardComponent implements OnInit {
  // 标记这三个Subject不需要自动unsubscribe
  @skipCheck()
  public widget$: ReplaySubject<DashboardWidget> = new ReplaySubject<DashboardWidget>(1);
  @skipCheck()
  public itemHistory$: ReplaySubject<DeviceHistoryItemEntry[]> = new ReplaySubject<DeviceHistoryItemEntry[]>(1);
  @skipCheck()
  public values$: ReplaySubject<TimeSeries[]> = new ReplaySubject<TimeSeries[]>(1);

  // 其余原有代码保持不变
}

方案2：改用untilDestroyed操作符管理订阅 #

关闭checkProperties配置，给所有显式的订阅流添加untilDestroyed(this)操作符，只取消订阅的观察者，不会关闭Subject本身。
代码示例：

@UntilDestroy() // 去掉checkProperties配置
@Component({
  selector: 'diagram-card',
  template: `<widget-diagram [data$]="values$"></widget-diagram>`,
  styleUrls: ['./diagram-card.component.scss'],
})
export class DiagramCardComponent implements OnInit {
  // 不需要加@skipCheck
  public widget$: ReplaySubject<DashboardWidget> = new ReplaySubject<DashboardWidget>(1);
  public values$: ReplaySubject<TimeSeries[]> = new ReplaySubject<TimeSeries[]>(1);

  private onUpdate(): void {
    this.widget$
      .pipe(
        debounceTime(500),
        distinctUntilChanged(),
        switchMap((widget) =>
          combineLatest([
            of(widget),
            this.store.select(DeviceHistoryStore.getItemHistoryEntries(this.deviceId, widget.data[0].itemId)),
          ])
        ),
        untilDestroyed(this) // 加操作符，组件销毁时自动取消该订阅
      )
      .subscribe(([widget, deviceHistory]) => {
        // 业务逻辑
        this.values$.next(diagram);
      });
  }

  // 其余所有subscribe的流都统一加untilDestroyed(this)操作符即可
}

方案3：调用next前增加状态校验（兜底方案） #

每次调用Subject的next方法前，先判断Subject的closed状态，仅当Subject未关闭时才发送事件，从调用侧避免报错。
代码示例：

.subscribe(([widget, deviceHistory]) => {
  // 先判断状态再执行next
  if (this.values$.closed) return;

  const diagram: TimeSeries[] = [];
  // 业务逻辑省略
  this.values$.next(diagram);
});

内容的提问来源于stack exchange，提问作者laprof