使用RxJS combineLatest过滤数据流时gridData始终为空的问题

问题现象 #

组件中使用combineLatest将过滤输入流与@Input传入的gridData Observable结合，但combineLatest回调中gridData始终为空数组。不过单独在模板中使用(gridData | async)可以正常获取所有记录。

组件代码如下：

export class RecordSearchComponent implements OnDestroy {
  @Input() gridData: Observable<TrainingRecord[]> = of([]);

  private filterSubject = new BehaviorSubject<string>('');
  private readonly debounceTimeMs = 300;

  private filterInput$ = this.filterSubject.pipe(
    debounceTime(this.debounceTimeMs),
    distinctUntilChanged(),
  )

  filteredData$ = combineLatest([this.filterInput$, this.gridData]).pipe(
    tap(([filter, data]) => console.log(filter, data)),
    switchMap(([filter, data]) => {
      if (filter === '' || filter === null) {
        return of(data);
      }
      const filterDescriptor = {
        field: 'nameFull',
        operator: 'contains',
        value: filter,
        ignoreCase: true,
      };
      return of(filterBy(data, filterDescriptor));
    })
  );

  ngOnDestroy(): void {
    this.filterSubject.complete();
  }

  onFilter() {
    this.filterSubject.next(this.inputText);
  }
}

原因分析 #

combineLatest的核心特性是：只有当所有传入的Observable都至少发射过一次值后，才会触发输出。这里：

• filterSubject是BehaviorSubject，初始化时会立刻发射空字符串，满足发射条件
• 如果gridData是冷Observable（比如HTTP请求返回的Observable），或者它的真实数据发射时机晚于filterInput$的第一次发射，那么combineLatest第一次触发时，gridData还没有发射过真实数据，就会使用定义时的默认值of([])，导致拿到空数组。
• 模板中的async管道会单独订阅gridData，当gridData发射真实数据时能正常获取，但这个订阅和combineLatest中的订阅是独立的，不会共享数据（冷Observable的特性）。

解决方案 #

方案1：给gridData添加startWith确保初始发射值 #

通过startWith操作符让gridData先发射一个空数组（或默认值），确保combineLatest能在gridData真实数据到来前就触发，后续真实数据发射时会自动更新：

filteredData$ = combineLatest([
  this.filterInput$,
  this.gridData.pipe(startWith([])) // 添加startWith
]).pipe(
  tap(([filter, data]) => console.log(filter, data)),
  switchMap(([filter, data]) => {
    if (filter === '' || filter === null) {
      return of(data);
    }
    const filterDescriptor = {
      field: 'nameFull',
      operator: 'contains',
      value: filter,
      ignoreCase: true,
    };
    return of(filterBy(data, filterDescriptor));
  })
);

方案2：改用withLatestFrom（优先响应过滤输入） #

如果只需要在过滤输入变化时获取gridData的最新值，可以改用withLatestFrom。它会在filterInput$发射值时，获取gridData的当前最新值，同样建议搭配startWith避免第一次过滤时无数据：

filteredData$ = this.filterInput$.pipe(
  withLatestFrom(this.gridData.pipe(startWith([]))),
  tap(([filter, data]) => console.log(filter, data)),
  switchMap(([filter, data]) => {
    // 原有过滤逻辑
    if (filter === '' || filter === null) {
      return of(data);
    }
    const filterDescriptor = {
      field: 'nameFull',
      operator: 'contains',
      value: filter,
      ignoreCase: true,
    };
    return of(filterBy(data, filterDescriptor));
  })
);

方案3：将gridData转为热Observable共享数据流 #

如果gridData是冷Observable（比如每次订阅都会重新发起请求），可以用shareReplay(1)让多个订阅（combineLatest和模板async）共享同一个数据流，避免重复请求，同时确保数据一致：

export class RecordSearchComponent implements OnDestroy {
  private _gridData$: Observable<TrainingRecord[]> = of([]);

  @Input() set gridData(data$: Observable<TrainingRecord[]>) {
    this._gridData$ = data$.pipe(shareReplay(1)); // 转为热Observable
  }

  private filterSubject = new BehaviorSubject<string>('');
  private readonly debounceTimeMs = 300;

  private filterInput$ = this.filterSubject.pipe(
    debounceTime(this.debounceTimeMs),
    distinctUntilChanged(),
  )

  filteredData$ = combineLatest([this.filterInput$, this._gridData$]).pipe(
    tap(([filter, data]) => console.log(filter, data)),
    switchMap(([filter, data]) => {
      // 原有过滤逻辑
      if (filter === '' || filter === null) {
        return of(data);
      }
      const filterDescriptor = {
        field: 'nameFull',
        operator: 'contains',
        value: filter,
        ignoreCase: true,
      };
      return of(filterBy(data, filterDescriptor));
    })
  );

  // 其余代码不变
}

总结 #

核心问题在于combineLatest的触发条件和gridData的Observable类型（冷/热）、发射时机。根据实际场景选择上述方案即可解决数据为空的问题。

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