Angular 8中RxJS Timer页面跳转后仍运行问题求助

问题描述 #

在Angular 8项目中实现了两个功能：

• 每30秒自动调用一次API，通过RxJS timer实现定时；
• 收到WebSocket推送的数据时，立即调用相同API，并将定时重置为30秒。

当前问题：页面跳转离开组件后，timer仍在后台运行，网络请求持续出现，怀疑是takeUntil操作符导致的订阅泄漏。

现有代码 #

import { Observable, Subject, timer } from 'rxjs';
import { finalize, takeUntil, takeWhile, tap } from 'rxjs/operators';

countdown = 30;
receivedWS = false;
private _unsubscribeAll: Subject<any> = new Subject();

ngOnInit() {
    this._networkService.getEventListener().pipe(takeUntil(this._unsubscribeAll)).subscribe((res) => {
      if (res) {
        this.receivedWS = true;
        this.myService.getUsers().then(() => this.receivedWS = false);
      }
    });

    this._timer();
}

private _timer(): void {
    timer(1000, 1000).pipe(
      finalize(() => {
        this.countdown = 30;
        this.myService.getUsers(); // API调用
        this._timer();
      }),
      takeWhile(() => this.countdown > 0 && this.receivedWS = false),
      takeUntil(this._unsubscribeAll),
      tap(() => this.countdown--)
    ).subscribe();
}

ngOnDestroy(): void {
    this._unsubscribeAll.next(true);
    this._unsubscribeAll.complete();
}

已尝试操作 #

曾尝试显式取消订阅Subject，导致出现UnsubscriptionErrorImpl错误，代码如下：

ngOnDestroy(): void {
    this._unsubscribeAll.next(true);
    this._unsubscribeAll.complete();
    this._unsubscribeAll.unsubscribe();
}

问题分析 #

1. 递归调用导致多订阅泄漏：_timer()方法在finalize中递归调用自身，每次执行都会创建新的timer订阅。组件销毁时，takeUntil只能终止当前活跃的timer，但之前递归生成的订阅可能仍在运行，甚至在组件销毁后，finalize触发时还会创建新的订阅。
2. takeWhile逻辑错误：this.receivedWS = false是赋值操作而非相等判断，导致takeWhile的条件逻辑完全错误，timer的运行行为不符合预期。
3. 操作符顺序错误：takeUntil未放在pipe的最后位置，当组件销毁触发takeUntil时，前面的finalize仍会执行，进而递归调用_timer()生成新订阅。
4. Subject操作错误：Subject.complete()已经会关闭Subject并通知所有订阅者，无需再调用unsubscribe()，这会导致后续操作触发UnsubscriptionErrorImpl。

解决方案 #

重构timer逻辑，使用switchMap管理订阅生命周期，避免递归调用，同时修正逻辑错误：

import { Observable, Subject, timer } from 'rxjs';
import { switchMap, takeUntil, tap, startWith, takeWhile } from 'rxjs/operators';

countdown = 30;
private _unsubscribeAll: Subject<any> = new Subject();
private _resetTimer$: Subject<void> = new Subject();

ngOnInit() {
    // WebSocket监听：接收数据时调用API并重置timer
    this._networkService.getEventListener().pipe(
      takeUntil(this._unsubscribeAll)
    ).subscribe((res) => {
      if (res) {
        this.myService.getUsers().then(() => {
          this._resetTimer$.next();
        });
      }
    });

    this._initTimer();
}

private _initTimer(): void {
    this._resetTimer$.pipe(
      startWith(null), // 初始触发一次，启动第一轮定时
      switchMap(() => {
        this.countdown = 30;
        // 每秒倒计时，直到countdown为0
        return timer(1000, 1000).pipe(
          tap(() => this.countdown--),
          takeWhile(() => this.countdown > 0)
        );
      }),
      // 倒计时结束后调用API，并触发下一轮定时
      tap(() => {
        this.myService.getUsers();
        this._resetTimer$.next();
      }),
      takeUntil(this._unsubscribeAll) // 放在最后，确保组件销毁时终止所有订阅
    ).subscribe();
}

ngOnDestroy(): void {
    this._unsubscribeAll.next(true);
    this._unsubscribeAll.complete();
    this._resetTimer$.complete();
}

重构优势 #

• 用switchMap管理订阅：每次重置timer时，自动取消上一轮的timer订阅，避免多订阅泄漏；
• 移除递归调用：通过_resetTimer$ Subject触发新一轮定时，逻辑更清晰，生命周期可控；
• 修正逻辑错误：去掉takeWhile中的赋值操作，使用正确的相等判断；
• 操作符顺序合理：takeUntil放在pipe最后，确保组件销毁时所有相关订阅立即终止；
• 避免Subject操作错误：仅调用complete()关闭Subject，无需额外unsubscribe()。

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