如何在Angular中使用RxJS正确实现链式接口请求？

核心问题 #

1. 附加请求D/E/F直接在流内部调用subscribe，脱离了主Observable控制流，主链触发回调时无法感知这三个请求的完成状态
2. 没有使用组合操作符对所有请求的完成信号做聚合，无法统一获取全部请求结束的时机

修正后代码 #

// 组件中增加加载状态控制变量
isLoading = false;
objects: IObjectCExtended[] = [];

private fetchData(): void {
  this.isLoading = true;
  this.backendService.getObjectA().pipe(
    // 展开A返回的数组，逐个处理每个ObjectA
    mergeMap((objects: IObjectA[]) => objects),
    // 对每个ObjectA发起请求B
    mergeMap((objectA: IObjectA) => this.backendService.getObjectB(objectA.id)),
    // 展开B返回的数组，逐个处理每个ObjectB
    mergeMap((objectsB: IObjectB[]) => objectsB),
    // 对每个ObjectB发起请求C
    mergeMap((objectB: IObjectB) => this.backendService.getObjectC(objectB.id)),
    // 处理每个ObjectC，并行发起D/E/F请求，全部返回后组装扩展对象
    mergeMap((objectC: IObjectC) => {
      const extendedObject: IObjectCExtended = objectC as IObjectCExtended;
      // 用forkJoin组合三个并行请求，全部完成后才会继续向下传递
      return forkJoin([
        this.backendService.getAdditionalInformationD(objectC.id),
        this.backendService.getAdditionalInformationE(objectC.id),
        this.backendService.getAdditionalInformationF(objectC.id)
      ]).pipe(
        // 把请求结果赋值到扩展对象上，返回最终的扩展对象
        map(([infoD, infoE, infoF]) => {
          extendedObject.additionalInfoD = infoD;
          extendedObject.additionalInfoE = infoE;
          extendedObject.additionalInfoF = infoF;
          return extendedObject;
        })
      )
    }),
    // 收集所有处理完成的扩展对象，等全部处理完后一次性返回数组
    toArray()
  ).subscribe({
    next: (allExtendedObjects: IObjectCExtended[]) => {
      this.objects = allExtendedObjects;
    },
    complete: () => {
      // 所有请求全部完成，关闭加载动画
      this.isLoading = false;
    },
    error: (err) => {
      // 错误处理，按需添加
      console.error('数据加载失败', err);
      this.isLoading = false;
    }
  })
}

UI 控制示例 #

<!-- 加载动画 -->
<div *ngIf="isLoading" class="loading-spinner"></div>
<!-- 业务内容，所有请求完成后才显示 -->
<div *ngIf="!isLoading" class="content">
  <!-- 渲染objects数组的内容 -->
</div>

关键操作符说明 #

• forkJoin：接收一组Observable，仅当所有Observable都完成时，返回包含所有结果的数组，适合处理并行的独立请求
• toArray：收集源流发出的所有值，等源流complete后一次性返回包含所有值的数组，确保拿到所有处理完成的对象
• 所有异步请求都被整合到主Observable流中，不会出现脱离控制的异步任务，complete回调触发时必然所有请求都已结束

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