rxjs - MergeMap 来自 Observables 数组

Question

TLDR:工作示例在这个问题的最后一个代码块中。查看 @bryan60 的答案，了解使用 concat 而不是 mergeMap 的工作示例。

---

我试图按顺序运行多个远程请求，但只执行了第一个可观察对象。

请求的数量各不相同，所以我无法采用将可观察对象相互嵌套的狡猾解决方案。

我正在使用以下代码:

const observables = [
  observable1,
  observable2,
  ...
];

from(observables).pipe(
  mergeMap(ob=> {
    return ob.pipe(map(res => res));
  }, undefined, 1)
).subscribe(res => {
  console.log('Huzzah!');
})

在过去(rxjs 5.5)我使用了以下内容:

let o = Observable.from(observables).mergeMap((ob) => {
  return ob;
}, null, 1);

o.subscribe(res => {
  console.log('Huzzah!');
})

我不确定我做错了什么，任何人都可以解释一下吗？

另一个请求是只打印“Huzzah!”在所有请求完成时执行一次，而不是针对每个单独的 Observable。

编辑:

从我的原始代码中删除 undefined 将使其正常工作，但是还有另一个问题导致仅执行第一个可观察对象。

我正在使用 Angular 的 HttpClient 进行远程请求。我的可观察代码如下所示:

const observables = [];

// Only the first observable would be executed
observables.push(this.http.get(urla));
observables.push(this.http.get(urlb));
observables.push(this.http.get(urlc));

将 .pipe(take(1)) 添加到每个 observable 会导致每个 observable 被执行:

const observables = [];

// All observables will now be executed
observables.push(this.http.get(urla).pipe(take(1));
observables.push(this.http.get(urlb).pipe(take(1));
observables.push(this.http.get(urlc).pipe(take(1));

我最终使用的代码按顺序执行所有可观察对象并且只触发一次 Huzzah! 是:

const observables = [];

observables.push(this.http.get(urla).pipe(take(1));
observables.push(this.http.get(urlb).pipe(take(1));
observables.push(this.http.get(urlc).pipe(take(1));

from(observables).pipe(
  mergeMap(ob=> {
    return ob.pipe(map(res => res));
  }, 1),
  reduce((all: any, res: any) => all.concat(res), [])
).subscribe(res => {
  console.log('Huzzah!');
})

感谢@bryan60 帮助我解决这个问题。

Answer 1

如果这些是已完成的 http 请求，我认为您的错误是由对删除结果选择器的 mergeMap 签名的更改引起的。在不知道您使用的是哪个版本的情况下很难确定，因为它在那里，然后被删除，然后又被添加，并且他们在 v7 中再次永久删除它。

如果你想按顺序运行它们......这就是你所需要的......

// concat runs input observables sequentially
concat(...observables).subscribe(res => console.log(res))

如果你想等到他们都完成发射，这样做:

concat(...observables).pipe(
  // this will gather all responses and emit them all when they're done
  reduce((all, res) => all.concat([res]), [])
  // if you don't care about the responses, just use last()
).subscribe(allRes => console.log(allRes))

在我的个人实用程序 rxjs 库中，我总是包含一个 concatJoin 运算符，它像这样组合 concat 和 reduce。

唯一的技巧是 concat 需要 observables 完成，直到它移动到下一个，但是 mergeMap 也是如此，并发订阅设置为 1。 .所以那应该没问题。像 http 请求这样的东西很好，因为它们在一次发射后自然完成。websockets 或主题或事件发射器的行为会有点不同，必须手动完成，或者使用像 first 或 这样的运算符采取或在源头。