Angular使用BehaviorSubject时forkJoin与combineLatest消费数据异常问题

问题原因 #

尝试1失败原因

• 同步返回of(profile)，forkJoin的异步回调还未执行，返回的永远是初始的空对象
• 服务内部直接调用subscribe没有做取消订阅处理，存在内存泄漏
• 本身forkJoin的逻辑是可行的：你的status$在加载完成后会主动调用complete，所有输入Observable都会触发完成事件，符合forkJoin的执行要求

尝试2失败原因

• RxJS v6及以上版本已经废弃了combineLatest的最后一个结果映射参数，必须通过pipe(map())处理返回值，这是你运行异常的核心原因
• 如果没有配置单次取值逻辑，后续任意输入Observable更新都会触发重复计算，不需要实时更新的场景会产生多余开销

最优实现方案 #

根据你的场景（一次性消费所有数据返回结果），推荐用forkJoin直接返回处理后的Observable，不要在服务内部订阅：

// DriverProfile.Service.ts 最终版
getDriverProfile(data:IDriverData): Observable<IDriverProfile>{
  return forkJoin({
    qualyTask: this.qualifyService.getDriverQualifyingLaps(data.driver),
    standingsTask: this.driverStandingsService.getDriverStandings(data.driver),
    resultsTask: this.resultService.getDriverResults(data.driver),
    raceTask: this.raceService.getAllRaces(),
    statusTask: this.statusService.status$,
    teamsTask: this.teamsService.getAllConstructors()
  }).pipe(
    map(res => {
      const calcResult = this.getResults(res.resultsTask, res.raceTask, res.standingsTask, res.statusTask, res.teamsTask);
      return {
        driverData: data,
        seasonsData: calcResult.seasons,
        careerStats: calcResult.stats
      } as IDriverProfile;
    })
  );
}

如果后续需要支持共享数据更新后自动重新计算IDriverProfile，可以用combineLatest的方案：

// 支持实时更新的版本
getDriverProfile(data:IDriverData): Observable<IDriverProfile>{
  return combineLatest([
    this.qualifyService.getDriverQualifyingLaps(data.driver),
    this.driverStandingsService.getDriverStandings(data.driver),
    this.resultService.getDriverResults(data.driver),
    this.raceService.getAllRaces(),
    this.statusService.status$,
    this.teamsService.getAllConstructors()
  ]).pipe(
    // 如果只需要获取一次结果，取消下面这行的注释
    // take(1),
    map(([qualy, standings, results, races, status, teams]) => {
      const calcResult = this.getResults(results, races, standings, status, teams);
      return {
        driverData: data,
        seasonsData: calcResult.seasons,
        careerStats: calcResult.stats
      } as IDriverProfile;
    })
  );
}

额外提示 #

当前你的Status.Service.ts里加载完数据后会调用_StatusSource.complete()，如果后续需要动态更新status数据，要删掉这个complete调用，这种场景下就不能用forkJoin，必须用combineLatest的方案。
组件侧使用时建议通过async管道订阅，自动处理订阅销毁，避免内存泄漏。

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