在forkJoin中使用valueChanges无法获取请求值的问题

原正常运行的forkJoin代码 #

const sources = [this.btApiClient.getAll(), this.buApiClient.getAll()];

forkJoin(sources)
  .pipe(
    map(([buildTypes, businessUnits]) => {
      this.bt$.next(buildTypes.result as GetAllBtsResponseItem[]);
      this.bu$.next(businessUnits.result as GetAllBusReponseItem[]);
    }),
    takeUntil(this.onDestroy$)
  ).subscribe()

需要加入的额外逻辑 #

原本独立运行的搜索框数据流处理：

this.tmSearcher.valueChanges.pipe(
  startWith(''),
  debounceTime(300),
  switchMap((searchStr) => {
    return this.npmApiClient.searchMember(searchStr ? searchStr : undefined, 100, 1, undefined, undefined);
  }),
  takeUntil(this.onDestroy$),
).subscribe(res => {
  this.teamMembers$.next(res!.result);
});

修改后的问题代码 #

将搜索逻辑封装为Test1方法并加入forkJoin，修改后代码：

const sources = [this.btApiClient.getAll(), this.buApiClient.getAll(), this.Test1()];

forkJoin(sources)
  .pipe(
    map(([buildTypes, businessUnits, tm]) => {
      this.bt$.next(buildTypes.result as GetAllBtsResponseItem[]);
      this.bu$.next(businessUnits.result as GetAllBusReponseItem[]);
    }),
    takeUntil(this.onDestroy$)
  ).subscribe()
}

private Test1() {
  return this.teamMembersSearcher.valueChanges.pipe(
    startWith(''),
    debounceTime(300),
    switchMap((searchStr) => {
      return this.npmApiClient.searchMember(searchStr ? searchStr : undefined, 100, 1, undefined, undefined);
    }),
    takeUntil(this.onDestroy$),
  )
}

问题现象 #

修改后三个HTTP请求均已发起，但forkJoin的map操作符和subscribe回调完全不执行，无法完成变量赋值。

原因分析 #

forkJoin的核心特性是：必须等待所有传入的Observable都完成（发出complete信号），才会合并发出最后一次值。
而teamMembersSearcher.valueChanges是一个持续数据流——只要搜索框内容变化，它就会持续发射新值，永远不会自动触发complete。因此forkJoin会一直等待这个Observable完成，永远不会执行后续逻辑。

解决方案 #

方案1：让搜索Observable只发射一次值并完成 #

如果只需要获取搜索框的初始值（对应startWith('')的请求结果），可以给Test1的Observable添加take(1)，强制它在获取第一个值后完成：

private Test1() {
  return this.teamMembersSearcher.valueChanges.pipe(
    startWith(''),
    debounceTime(300),
    switchMap((searchStr) => {
      return this.npmApiClient.searchMember(searchStr || undefined, 100, 1, undefined, undefined);
    }),
    take(1), // 仅取第一个值，触发Observable完成
    takeUntil(this.onDestroy$)
  );
}

方案2：分开处理一次性请求与持续数据流 #

如果业务需要监听搜索框后续变化并重新请求数据，那么forkJoin不适合处理这个持续数据流，应该将一次性并行请求和搜索数据流分开订阅：

// 处理一次性并行请求
const sources = [this.btApiClient.getAll(), this.buApiClient.getAll()];
forkJoin(sources)
  .pipe(takeUntil(this.onDestroy$))
  .subscribe(([buildTypes, businessUnits]) => {
    this.bt$.next(buildTypes.result as GetAllBtsResponseItem[]);
    this.bu$.next(businessUnits.result as GetAllBusReponseItem[]);
  });

// 单独处理搜索框的持续数据流
this.teamMembersSearcher.valueChanges.pipe(
  startWith(''),
  debounceTime(300),
  switchMap((searchStr) => {
    return this.npmApiClient.searchMember(searchStr || undefined, 100, 1, undefined, undefined);
  }),
  takeUntil(this.onDestroy$)
).subscribe(res => {
  this.teamMembers$.next(res!.result);
});

这种方案更贴合实际业务场景，既能初始化加载基础数据，又能响应后续的搜索交互。

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