是否有一个操作符可以用作 concatMap 但具有多个内部可观察对象

Question

我正在使用 observable 来查询我的数据库。 这个 observable 将返回一个包含所有找到的匹配对象的数组。 我的问题是我想用我将从另一个 API 检索到的更多细节来映射 observable。

我尝试了 concatMap 但它只让我在初始 observable 中嵌套 1 个 observable

const usersAPI$: Observable<Users[]> = //  something
const assistsAPI$: Observable<Assists[]> = //  something

const modifiedAssists$ = assistsAPI$.find().pipe(
  concatMap(assists =>
    assists.map(assist => usersAPI$.get(assist.userID)
       .pipe(
          map(user => 
            assist = {...assist}, ...{userName: user.userName}
          )
        )
    )
  )
);

您可以在这里看到一个类似的工作示例，在 stackblitz 上， https://stackblitz.com/edit/rxjs-concatmap-issue-stackoverflow 其中“结果”是使用 concatMap 和 “result2”是我期望工作的非工作方式

Answer 1

您必须以某种方式处理您的内部 Observable 数组。您可以根据需要使用forkJoin、merge 或concat。

forkJoin

forkJoin 将并行处理您的内部 API 调用，并在所有 API 调用完成后返回一个数组。请注意，内部的 Observables 必须完成。我认为这应该适合您的应用。

import { forkJoin } from 'rxjs';

const modifiedAssists$ = assistsAPI$.find().pipe(
  concatMap(assists => forkJoin(assists.map(assist => someObservable ))
);

合并

merge 将立即订阅所有内部 API 调用，并在所有调用到达时一一发出结果。

import { merge } from 'rxjs';

const modifiedAssists$ = assistsAPI$.find().pipe(
  concatMap(assists => merge(...assists.map(assist => someObservable )),
  // toArray() add the toArray operator if you need the result as an array
);

这将为您的 stackblitz 提供所需的结果，但我认为您的 stackblitz 有点误导，而不是您在问题的示例代码中寻找的内容，因为您的 stackblitz 中的内部 Observable 多次发出并且最终输出不是数组。如果内部请求的顺序无关紧要，它们都发出一个值然后完成，您需要将所有请求的结果作为一个数组，只需使用forkJoin。

连接

concat 将一个接一个地订阅所有内部 API 调用。序列中的下一个 Observable 只有在前一个完成后才会被订阅。因此，执行将比 forkJoin 或 merge 慢，因为下一个 HTTP 请求只会在前一个返回值之后执行。如果对您的 userAPI 的调用必须按照与数组中的 assists 相同的顺序进行，请使用此选项。

import { concat } from 'rxjs';

const modifiedAssists$ = assistsAPI$.find().pipe(
  concatMap(assists => concat(...assists.map(assist => someObservable )),
  // toArray() add the toArray operator if you need the result as an array
);

Answer 2

您可以使用combineLatest()（作为静态非管道导入）

import {combineLatest} from 'rxjs';

const usersAPI$: Observable<Users[]> = //  something
const assistsAPI$: Observable<Assists[]> = //  something

const combinedObservable$ = combineLatest(usersAPI$,assistsAPI$, someOtherStuff$);

const modifiedAssists$ = combinedObservable$.pipe(
   map([usersApiValues, assistsAPIValues, someOtherStuffValues] => { 
      /** what you want here */
   }) 
);

只要考虑到combineLatest 中的每个 Observables 应该至少发出一次（例如，您可以用一个空数组开始它们）

Answer 3

我相信你想要mergeMap？

它将外部可观察对象映射到每个外部发射的已解析内部可观察对象。

const result = clicks.pipe(
  concatMap(ev =>
    interval(1000).pipe(take(4))
  )
);

const result2 = clicks.pipe(
  mergeMap(ev =>
    arr.map(() => interval(1000).pipe(take(4)))
  )
);
result.subscribe(x => {
  console.log(x)
});

result2.subscribe(x => {
  console.log(x)
});