在对每个响应执行某些操作时并行执行 observable答案

【问题标题】：Execute observable in parallel while do something with each response在对每个响应执行某些操作时并行执行 observable
【发布时间】：2019-06-16 21:05:49
【问题描述】：

我有一个可观察的列表 (obs$)。

 obs$ = [ob1$, ob2$..];

每次每个 observable(ob$) 完成时，

我需要做点什么，例如。分别更新与每个可观察对象相关的本地对象 ob1$ = ob1(v1).do_something(); ob1$.subscribe(e => v1.state = 'ok');

要处理 obs$ 中的所有 ob$ ，我可以递归地遍历所有 ob$ 并完成我想做的事情 - 更新 v1、v2...

现在，当我考虑如何并行处理时，考虑使用forkJoin

Observable.forkJoin(...obs$).subscribe( res => ..)

联合响应 (res) 不适合我的情况。我对么？

Rxjs 中有什么方法可以让我实现我想要的吗？

【问题讨论】：

为什么您认为Observable.forkJoin(...obs$).subscribe( res => ..) 不适合您的情况。 ?
如果您的可观察对象的执行顺序对您来说并不重要，那么您可能需要考虑mergeMap 运算符，因为它会在您从每个内部可观察对象获得结果时执行。这可能会有所帮助：blog.angularindepth.com/…
不适合，因为只要不是错误，响应就无关紧要。所以我需要知道每个人是否都没有错误。至于顺序，没关系。这就是我尝试并行的原因
就像 dee zg 所说：常规的 mergeMap 就是这样做的。只需将您的 x 次调用排入队列，只要响应进来，它们就会被处理。但请注意：JS 中没有真正的并行执行。您可以等待 x 次 api 调用，但处理这些响应永远不会真正并行发生。

标签： angular rxjs

【解决方案1】：

您可以将每个响应映射到 index 或其他任何内容，这样您就知道哪个源 Observable 完成了，然后只需使用 merge：

obs$ = [ob1$, ob2$, ...];
const mappedObs$ = obs$.map((ob$, index) => ob$.pipe(
  map(value => ({ index, value })),
));

merge(...mappedObs$).subscribe(({ index, value }) => {
  if (index === 42) {
    // do whatever
  }
});

【讨论】：

或者只是在您添加到每个可观察的obs$ = [obs1$.pipe(map(value => doSomething1(value))), obs2$.pipe(...), ...]; merge(obs$).subscribe() 的map 中完成工作