自定义RxJS buffer操作符导致Angular应用无法进入稳定状态如何解决？

问题根因 #

1. 重复订阅源Observable：你的bufferAll实现中，对同一个源Observable生成了3次独立订阅：
   • 给size$ = source$.pipe(bufferCount(bufferSize))的订阅
   • 给timeSpan$ = source$.pipe(bufferTime(bufferTimeSpan * 1000))的订阅
   • 给最后return source$.pipe(bufferFrom(...))的订阅
     重复订阅不仅会导致额外的性能开销，还会让bufferTime生成的定时器被多次初始化，始终有活跃的定时器任务被NgZone跟踪。
2. 定时器始终在NgZone内运行：即便你把订阅逻辑放在runOutsideAngular中，每次restart.next()触发switchMap重新订阅bufferTime流时，新生成的定时器默认还是会注册到当前的Angular Zone中，NgZone会认为始终有未完成的异步任务，导致应用永远无法进入稳定状态。
3. 触发逻辑冗余错误：用bufferCount/bufferTime的输出作为buffer操作符的触发信号属于逻辑冗余，这两个操作符本身就会缓存一份源数据，你实际只需要它们的"发射时机"作为触发信号，不需要它们缓存的数据，额外的缓存也会增加内存开销。

修复方案 #

第一步：重写自定义操作符，避免重复订阅和逻辑冗余 #

import { Observable, Subscription, timer, merge } from 'rxjs';
import { buffer, tap, scan, filter, share } from 'rxjs/operators';
import { NgZone } from '@angular/core';

export function bufferAll<T>(
  bufferTimeSpan: number, 
  bufferSize: number, 
  trigger$: Observable<any>,
  zone?: NgZone
): (source$: Observable<T>) => Observable<T[]> {
  return (source$) => {
    // 用share做多播，保证所有操作符只订阅一次源
    const sharedSource$ = source$.pipe(share());

    // 容量触发流：计数达到bufferSize时触发
    const sizeTrigger$ = sharedSource$.pipe(
      scan((count) => count + 1, 0),
      filter(count => count === bufferSize)
    );

    // 时间触发流：每隔bufferTimeSpan触发一次，支持Zone外运行
    const timeTrigger$ = new Observable(subscriber => {
      let timerSub: Subscription;
      const runTimer = () => {
        timerSub = timer(bufferTimeSpan * 1000).subscribe(() => {
          subscriber.next(void 0);
          runTimer();
        });
      };

      if (zone) {
        zone.runOutsideAngular(runTimer);
      } else {
        runTimer();
      }

      return () => timerSub?.unsubscribe();
    });

    // 合并三类触发信号
    const bufferTrigger$ = merge(sizeTrigger$, timeTrigger$, trigger$);

    return sharedSource$.pipe(
      buffer(bufferTrigger$),
      filter(items => items.length > 0)
    );
  };
}

第二步：调整使用方式 #

this.zone.runOutsideAngular(() => {
  const routeChanges$ = this.router.events.pipe(
    filter(event => event instanceof NavigationEnd)
  );

  this.dataSubject.asObservable().pipe(
    // 传入NgZone实例，确保定时器完全运行在Zone外
    bufferAll(60, 10, routeChanges$, this.zone)
  ).subscribe((data) => {
    this.zone.run(() => console.log(data));
  });
});

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