如何从内部Observable获取数据？RxJS嵌套流处理问题

现有代码如下：

this.cardDynamicFields$
  .pipe(
    filter((data) => !!data?.objectfieldsdescription),
    map((data) => data.objectfieldsdescription),
    mergeMap((fields) =>
      fields.map((field: any) => {
        const { iconClassPropName, classifierId } = field;
        return this.classificatorService
          .getClassifierRows(classifierId)
          .pipe(
            map((data) => data.find((el) => el.id === iconClassPropName))
          );
      })
    )
  )
  .subscribe((e) => {
    console.log(e);
  });

其中this.classificatorService.getClassifierRows(classifierId)会返回一个内部Observable。使用mergeMap处理后，subscribe中始终得到Observable数组，请问如何获取该内部Observable的结果？

问题出在mergeMap的返回值上：你现在返回的是fields.map()生成的Observable数组，但mergeMap需要接收单个Observable，或者发出Observable的Observable（Observable<Observable<T>>），而非数组本身。需要把Observable数组转换成能发出实际数据的Observable，以下是两种常用处理方式：

方式1：用forkJoin并行请求，一次性获取所有结果 #

适合需要等待所有请求完成后，统一拿到所有结果的场景：

this.cardDynamicFields$
  .pipe(
    filter((data) => !!data?.objectfieldsdescription),
    map((data) => data.objectfieldsdescription),
    mergeMap((fields) =>
      // 用forkJoin包裹Observable数组，等待所有请求完成后返回结果数组
      forkJoin(
        fields.map((field: any) => {
          const { iconClassPropName, classifierId } = field;
          return this.classificatorService
            .getClassifierRows(classifierId)
            .pipe(
              map((data) => data.find((el) => el.id === iconClassPropName))
            );
        })
      )
    )
  )
  .subscribe((results) => {
    // results是所有内部Observable返回值组成的数组
    console.log(results);
  });

方式2：用merge逐个获取结果 #

适合不需要等待所有请求完成，每个请求结束就立即拿到对应结果的场景：

import { merge } from 'rxjs';

this.cardDynamicFields$
  .pipe(
    filter((data) => !!data?.objectfieldsdescription),
    map((data) => data.objectfieldsdescription),
    mergeMap((fields) =>
      // 用merge把Observable数组转换成单个Observable，逐个发出每个请求的结果
      merge(
        ...fields.map((field: any) => {
          const { iconClassPropName, classifierId } = field;
          return this.classificatorService
            .getClassifierRows(classifierId)
            .pipe(
              map((data) => data.find((el) => el.id === iconClassPropName))
            );
        })
      )
    )
  )
  .subscribe((singleResult) => {
    // 每个内部Observable完成时，都会触发这里并返回单个结果
    console.log(singleResult);
  });

核心原因说明 #

• fields.map()生成的是Observable数组，但这个数组本身不是Observable，mergeMap会直接把数组作为值传递给下游，所以subscribe里拿到的是Observable数组。
• forkJoin/merge可以将Observable数组转换成发出实际数据的Observable，mergeMap会订阅这个Observable，进而把内部每个请求的结果传递到subscribe中。

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