Angular RxJs中如何批量调用API更新ArrayList对象属性？

我有一个包含多个对象的ArrayList，需要先调用第一个API获取结果集，再为结果集中的每个对象调用第二个API获取详情并更新该对象属性。尝试用ForkJoin和forEach实现，但所有请求还未完成程序就返回了。

组件方法代码 #

fetchReleaseDtl(strProjName: string) {
    let searchParams = new HttpParams();
    searchParams = searchParams.append('api-version', '7.0');
    console.log('about to fetch release details for ' + strProjName);
    return this.http
      .get<APIRlsDtlResp>(
        'https://vsrm.dev.azure.com/XXX/' + strProjName + '/_apis/release/releases',
        {
          headers: new HttpHeaders({
            'Authorization': 'Basic XXX'
          }),
          params: searchParams,
          responseType: 'json'
        }
      )
      .pipe(
        map(responseData => {
          console.log(responseData);
          let RlsDtlArray: APIRlsDtl[] = responseData.value;
          let RlsDtlDisp: APIRlsDtlDisp[] = [];
          for (let i = 0; i < RlsDtlArray.length; i++) {
            if (RlsDtlArray[i].releaseDefinition.name != '' && RlsDtlArray[i].description != ''
              && RlsDtlDisp.filter(e => e.name === RlsDtlArray[i].releaseDefinition.name
                && e.description === RlsDtlArray[i].description).length == 0) {
              var id = RlsDtlArray[i].id;
              var RlsDtl: APIRlsDtlDisp = {
                id: RlsDtlArray[i].id,
                name: RlsDtlArray[i].releaseDefinition.name,
                status: RlsDtlArray[i].status,
                description: RlsDtlArray[i].description.slice(0, 50),
                duration: 0
              };
              RlsDtlDisp.push(RlsDtl);
            }
          }
          return RlsDtlDisp;
        }),map((result) => {
          forkJoin(result.map(a=>{
            this.http.get<APIRlsExecDtl>('https://vsrm.dev.azure.com/XXX/XXX/_apis/release/releases/' + a.id, {
              headers: new HttpHeaders({
                'Authorization': 'Basic XXX='
              }),
              params: searchParams,
              responseType: 'json'
            }).subscribe((resp)=>{
                console.log(resp.environments[0].timeToDeploy);
                return result;
            })
          })).subscribe((b)=>{
            console.log('From Fork Join');
            console.log(b);
            console.log('About to return to component');
            return result;
          })
          /*result.forEach(element => {
            this.http.get<APIRlsExecDtl>('https://vsrm.dev.azure.com/XXX/XXX/_apis/release/releases/' + element.id, {
              headers: new HttpHeaders({
                'Authorization': 'Basic XXX'
              }),
              params: searchParams,
              responseType: 'json'
            }).subscribe((resp) => {
              console.log('Received resp');
              console.log(resp.environments[0].status);
              console.log(resp.environments[0].timeToDeploy);
              element.status = resp.environments[0].status;
              element.duration = parseInt(resp.environments[0].timeToDeploy);
            });
          })*/
          
        }),
        catchError(errorRes => {
          // Send to analytics server
          console.log(errorRes);
          return throwError(errorRes);
        })
      );
  }

数据模型代码 #

export interface APIRlsDtl{
    id:number;
    name:string;
    status:string;
    createdOn:string;
    modifiedOn:string;
    description:string;
    modifiedBy:{
      displayName:string
    };
    createdBy:{
      displayName:string
    };
    releaseDefinition:{
      name:string
    };
  }
  export interface APIRlsDtlResp{
    count:number;
    value:APIRlsDtl[]
  }
  export interface APIRlsDtlDisp{
    id:number;
    name:string;
    description:string;
    status:string;
    duration:number;
  }
  export interface RlsExecDtl{
    status:string
    timeToDeploy:string
  }
  export interface APIRlsExecDtl{
    id:number;
    environments:[RlsExecDtl]
  }

原代码核心问题是在map操作符内部直接订阅了forkJoin和HTTP请求，map仅负责同步转换数据，无法处理异步嵌套请求，导致外层Observable不会等待内部异步请求完成就直接返回初始的result数组。

组件方法代码 #

fetchReleaseDtl(strProjName: string) {
    let searchParams = new HttpParams();
    searchParams = searchParams.append('api-version', '7.0');
    console.log('about to fetch release details for ' + strProjName);
    
    return this.http
      .get<APIRlsDtlResp>(
        `https://vsrm.dev.azure.com/XXX/${strProjName}/_apis/release/releases`,
        {
          headers: new HttpHeaders({
            'Authorization': 'Basic XXX'
          }),
          params: searchParams
        }
      )
      .pipe(
        // 过滤并生成初始展示数组
        map(responseData => {
          console.log(responseData);
          const RlsDtlArray: APIRlsDtl[] = responseData.value;
          const RlsDtlDisp: APIRlsDtlDisp[] = [];
          
          for (const item of RlsDtlArray) {
            // 用some替代filter提升性能
            if (item.releaseDefinition.name !== '' && item.description !== '' && 
                !RlsDtlDisp.some(e => e.name === item.releaseDefinition.name && e.description === item.description)) {
              const rlsDtl: APIRlsDtlDisp = {
                id: item.id,
                name: item.releaseDefinition.name,
                status: item.status,
                description: item.description.slice(0, 50),
                duration: 0
              };
              RlsDtlDisp.push(rlsDtl);
            }
          }
          return RlsDtlDisp;
        }),
        // 使用switchMap处理嵌套异步请求
        switchMap((resultArray) => {
          // 为每个元素生成详情请求Observable
          const detailObservables = resultArray.map(item => {
            return this.http.get<APIRlsExecDtl>(
              `https://vsrm.dev.azure.com/XXX/XXX/_apis/release/releases/${item.id}`,
              {
                headers: new HttpHeaders({
                  'Authorization': 'Basic XXX='
                }),
                params: searchParams
              }
            ).pipe(
              // 在请求回调中更新元素属性
              map(resp => {
                item.status = resp.environments[0].status;
                item.duration = parseInt(resp.environments[0].timeToDeploy);
                return item;
              })
            );
          });
          
          // forkJoin等待所有请求完成，返回更新后的数组
          return forkJoin(detailObservables);
        }),
        catchError(errorRes => {
          console.log(errorRes);
          return throwError(() => errorRes);
        })
      );
  }

优化后的数据模型代码 #

// 发布详情接口
export interface APIRlsDtl{
    id: number;
    name: string;
    status: string;
    createdOn: string;
    modifiedOn: string;
    description: string;
    modifiedBy: {
      displayName: string
    };
    createdBy: {
      displayName: string
    };
    releaseDefinition: {
      name: string
    };
}

// 发布详情响应接口
export interface APIRlsDtlResp{
    count: number;
    value: APIRlsDtl[]
}

// 用于展示的发布详情接口
export interface APIRlsDtlDisp{
    id: number;
    name: string;
    description: string;
    status: string;
    duration: number;
}

// 发布执行详情接口
export interface RlsExecDtl{
    status: string;
    timeToDeploy: string;
}

// 发布执行详情响应接口
export interface APIRlsExecDtl{
    id: number;
    environments: RlsExecDtl[];
}

1. 替换map为switchMap：switchMap会订阅内部的forkJoinObservable，确保外层Observable等待所有内部请求完成后再发射最终结果。
2. 移除内部订阅：所有HTTP请求返回Observable，由forkJoin统一管理，避免嵌套订阅导致的异步流程失控。
3. 直接更新原对象：在每个详情请求的map操作中更新元素属性，forkJoin完成后直接返回更新后的数组。
4. 优化数组过滤逻辑：用Array.some()替代filter(...).length == 0，提升性能。

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