如何使用RxJS调用嵌套依赖API并组装指定结果

问题背景 #

我有3个存在依赖关系的API：

• API 1：返回一个对象，包含GalleryImages字段，其Items字段为字符串ID数组。
• API 2：需传入API 1返回的字符串ID调用，返回的对象包含Items字段，该字段是对象数组，每个对象含AssetIds字符串ID数组。
• API 3：需传入API 2返回的Items中每个元素的ID列表调用，返回对象的Items字段为含URL字段的对象数组。

需要用RxJS操作符组合调用这些API，生成包含以下字段的对象：

• Title：字符串，取自API 1的title字段
• Images：URL数组，取自API 3的结果

我自己的实现存在空值处理不当等缺陷，寻求正确的RxJS组合方案。

初始实现代码 #

public getGalleryImagesContentDtoObservableByIds(ids: string[]): Observable<any[]> {

    const obs = new Subject<ImageGalleryComponentDto[]>();
    const listResult: ImageGalleryComponentDto[] = [];

    if (ids && ids.length > 0) {
      this.galleryImagesService.queryGalleryImagesContent({ ids: ids.join(",") }).subscribe(a => {
        a.items.forEach(b => {
          const listGalleryImage = b.data.Images;

          if (listGalleryImage && listGalleryImage['iv'] && listGalleryImage['iv'].length > 0) {
            this.galleryImageService.queryGalleryImageContent({ ids: b.data.Images['iv'].join(",") }).subscribe(c => {

              const item = new ImageGalleryComponentDto();
              item.title = b.data.Title[this.languagueService.languague];
              item.id = b.id;
              item.tag = b.data.Tag[this.languagueService.languague];

              const listImageDtoObs = c.items.map(d => {
                const listImage = d.data.Image;

                return this.assetService.assetsGetAssets({ ids: (d.data.Image && d.data.Image['iv'] &&d.data.Image['iv'].length > 0) ? d.data.Image['iv'].join(",") : "" , app: this.applicationConfig.appName }).pipe(map(e => {
                  const imageDtoItem = new ImageDto();
                  imageDtoItem.title = d.data.Title[this.languagueService.languague];
                  imageDtoItem.description = d.data.Description[this.languagueService.languague];
                  const imageUrl = this.assetUtilityService.getLinkFromAssetDto(e);
                  imageDtoItem.imageUrl = imageUrl && imageUrl.length > 0 ? imageUrl[0] : "";

                  return imageDtoItem;
                }), catchError(err => of(err)))

                forkJoin(listImageDtoObs).subscribe(f => {
                  item.images = f;
                  listResult.push(item);
                  obs.next(listResult);
                })


              })
            })
          }
        })
      })
    }

    return obs;
  }

2022年7月29日更新：改进后的实现代码 #

public getGalleryImagesContentDtoObservableById(id: string): Observable<ImageGalleryComponentDto>{

    if(id){
      const obs =  this.galleryImagesService.getGalleryImagesContent({id: id})
      .pipe(
        mergeMap((galleryImagesRes)=>{
          return this.galleryImageService.queryGalleryImageContent({ids: galleryImagesRes.data.Images['iv'].join(",")}).pipe(map(a => {
            return {
              galleryImagesRes: galleryImagesRes,
              galleryImageRes: a.items
            }
          }))
        }),
        mergeMap((galleryImageRes)=>{

          const imageAssetIds = galleryImageRes.galleryImageRes.map(b =>{
            return b.data.Image[this.languagueService.languague][0];
          });

          const listImageDtp =  imageAssetIds.map(c =>{
            return this.assetService.assetsGetAsset({id:c, app: this.applicationConfig.appName}).pipe(map(d =>{
              const imageDto = new ImageDto();
              imageDto.imageUrl = this.assetUtilityService.getLinkFromAssetDto(d);
              return imageDto;
            }))
          })

          return forkJoin(listImageDtp).pipe(map(e =>{
            return {
              galleryImagesRes: galleryImageRes.galleryImagesRes,
              galleryImageRes: galleryImageRes.galleryImageRes,
              imagesDto: e
            }
          }))


        }
        )
      ).pipe(map(finalRes =>{
        const imageGalleryComponentDto = new ImageGalleryComponentDto();
        imageGalleryComponentDto.id = finalRes.galleryImagesRes.id;
        imageGalleryComponentDto.title = finalRes.galleryImagesRes.data.Title[this.languagueService.languague];
        imageGalleryComponentDto.images = finalRes.imagesDto;
        return imageGalleryComponentDto;
      }));

      return obs;
    }

    return of();
  }

正确的RxJS实现方案 #

下面的实现重点优化了空值校验、异常捕获，全程使用RxJS操作符避免嵌套订阅，保证数据流的纯净性：

import { Observable, of, forkJoin } from 'rxjs';
import { mergeMap, map, catchError, filter } from 'rxjs/operators';

public getGalleryImagesContentDtoObservableById(id: string): Observable<ImageGalleryComponentDto | null> {
  // 空ID直接返回null的Observable
  if (!id) {
    return of(null);
  }

  return this.galleryImagesService.getGalleryImagesContent({ id })
    .pipe(
      // 校验API1返回结果及必要字段
      filter(res => res?.data?.Images?.['iv']?.length > 0),
      mergeMap(galleryImagesRes => {
        const imageIds = galleryImagesRes.data.Images['iv'].join(',');
        // 调用API2，同时保留API1的结果
        return this.galleryImageService.queryGalleryImageContent({ ids: imageIds })
          .pipe(
            map(galleryImageRes => ({ galleryImagesRes, galleryImageItems: galleryImageRes.items })),
            // 捕获API2调用异常，返回空数据避免流中断
            catchError(() => of({ galleryImagesRes, galleryImageItems: [] }))
          );
      }),
      mergeMap(({ galleryImagesRes, galleryImageItems }) => {
        // 提取有效AssetId，过滤空值
        const validAssetIds = galleryImageItems
          .map(item => item?.data?.Image?.[this.languagueService.languague]?.[0])
          .filter(id => !!id);

        // 无有效ID时直接返回空图片数组
        if (validAssetIds.length === 0) {
          return of({ galleryImagesRes, imageDtos: [] });
        }

        // 批量调用API3，用forkJoin并行处理
        const imageObservables = validAssetIds.map(assetId =>
          this.assetService.assetsGetAsset({ id: assetId, app: this.applicationConfig.appName })
            .pipe(
              map(assetRes => {
                const imageUrl = this.assetUtilityService.getLinkFromAssetDto(assetRes);
                const imageDto = new ImageDto();
                imageDto.imageUrl = imageUrl?.length > 0 ? imageUrl[0] : '';
                return imageDto;
              }),
              // 单个API3调用失败时返回空ImageDto，不影响整体
              catchError(() => of(new ImageDto()))
            )
        );

        return forkJoin(imageObservables)
          .pipe(
            map(imageDtos => ({ galleryImagesRes, imageDtos }))
          );
      }),
      map(({ galleryImagesRes, imageDtos }) => {
        const dto = new ImageGalleryComponentDto();
        dto.id = galleryImagesRes.id;
        dto.title = galleryImagesRes.data.Title?.[this.languagueService.languague] || '';
        dto.images = imageDtos;
        return dto;
      }),
      // 全局捕获异常，返回null
      catchError(() => of(null))
    );
}

关键优化点 #

• 避免嵌套订阅：全程使用mergeMap、forkJoin等操作符组合数据流，替代手动订阅嵌套
• 严格空值校验：通过filter、可选链?.、filter(id => !!id)过滤所有可能的空值节点
• 异常隔离处理：每个API调用都添加catchError，单个请求失败不中断整个数据流，返回兜底值
• 数据流纯净性：不再手动创建Subject，直接通过操作符组合生成最终Observable
• 资源高效利用：用forkJoin并行调用API3，提升请求效率

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