angular - RXJS 在管道内组合多个可观察对象

Question

我有一个API调用，返回一定数量的ID。
这些 id 中的每一个都用于进行新的 api 调用。这些 API 调用的结果需要组合成一个对象。

起初，我在第一个 api 调用的 .pipe(map) 运算符中使用了一个循环。在此循环中，我进行了第二次api调用，并在每个调用中的.pipe(map)运算符中，都在 Angular 组件中编辑了一个变量。

这不是很漂亮，我实际上想知道这是否是线程安全的。我知道 javascript 是单线程的，但是让多个异步进程混淆同一个全局变量似乎不太安全。

之后，我只是通过循环 apiCall1 返回的 Id 将第二个 api 调用返回的 observable 存储在数组中，并使用 forkJoin 订阅并相应地处理每个结果(参见示例)。

然而，这不是很漂亮，我想知道是否有我可以在我的管道中使用的运算符？

因此，而不是(pseudocode):

  .pipe(
      map(ids=> {

        let observables = []
        for (const id of ids) {
         observables.push(this.service.getSomeStuff(id));
        }

        forkJoin(...observables).subscribe(dataArray) => {
          for (data of dataArray) {
            //Do something
          }
        });

      }),
      takeWhile(() => this.componentActive),
      catchError(error => {
        console.log(error);
        return throwError(error);
      })
    )
    .subscribe();

是否有一个运算符使它像这样:

  .pipe(
      map(ids=> {

        let observables = []
        for (const id of ids) {
         observables.push(this.service.getSomeStuff(id));
        }

      return observables
      }),
      forkJoin(dataArray => {
          for (data of dataArray) {
            //Do something
          }
        });
      takeWhile(() => this.componentActive),
      catchError(error => {
        console.log(error);
        return throwError(error);
      })
    )
    .subscribe();

Answer 1

这是你可以做的:

sourceObservable$.pipe(
  // depends on your need here you can use mergeMap as well
  switchMap(ids => {
    const observables = ids.map(i => this.service.getSomeStuff(id));
    return forkJoin(observables);
  }),
  tap(joined => {
    // joined will be an array of values of the observables in the same
    // order as you pushed in forkJoin
    for (data of joined) {
      // do something
    }
  }),
  takeWhile(() => this.componentActive),
  catchError(error => {
    console.log(error);
    return throwError(error);
  })
)
.subscribe();