如何并行调用所有Observable并将汇总结果存储到指定对象中

核心思路 #

• 你原代码串行执行的原因是在循环内用await等待单个请求完成再执行下一次循环，改用RxJS的forkJoin操作符即可实现所有请求并行发起，等待全部完成后统一处理结果
• 每个请求需要携带对应的层级（level）和指标键（sensorKey）信息，避免返回结果乱序后无法匹配原对象结构
• 单个请求返回后提前做total求和处理，所有请求完成后直接拼装成目标结构赋值即可

完整实现代码 #

import { forkJoin, catchError, of, map } from 'rxjs';

// 1. 生成所有请求的Observable数组
const requestList = [];
Object.keys(levels).forEach(level => {
  Object.keys(levels[level]).forEach(sensorKey => {
    const sensorId = levels[level][sensorKey];
    // 每个请求处理完返回对应的层级、键、求和结果
    const req$ = this.sensorService.drillData({ ...this.params, sensor: sensorId }).pipe(
      // 计算当前指标的total总和
      map((sensorData: SensorData[]) => {
        const total = sensorData.reduce((sum, item) => sum + (item.total || 0), 0);
        return { level, sensorKey, total };
      }),
      // 单个请求出错不影响其他请求，可根据需求调整默认值
      catchError(err => {
        console.error(`请求${level}的${sensorKey}指标失败:`, err);
        return of({ level, sensorKey, total: 0 });
      })
    );
    requestList.push(req$);
  });
});

// 2. 并行执行所有请求，全部完成后统一赋值
forkJoin(requestList).subscribe(resultList => {
  // 重置levelValues避免旧数据干扰
  this.levelValues = {
    level30: {},
    level15: {},
    level10: {},
    level5: {},
  };
  // 遍历结果填充到对应位置
  resultList.forEach(({ level, sensorKey, total }) => {
    this.levelValues[level as keyof SoilLevels][sensorKey] = total;
  });
  // 此处可添加赋值完成后的后续业务逻辑
});

补充说明 #

• forkJoin 会等待所有传入的Observable都发出完成信号后，才会返回所有结果的数组，所有请求并行发起，不会互相等待
• 单个请求的错误处理在catchError中完成，默认返回0作为失败后的默认值，你可以根据业务需求调整为null或者其他取值
• 结果返回后按照携带的level和sensorKey赋值，完全保留原levels的结构，符合你的需求

内容的提问来源于stack exchange，提问作者Guru Prakash Ram