来自观测阵列的MergeMap

Question

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

我试图按顺序运行许多远程请求，但只有第一个可观察到的请求被执行。

请求的数量各不相同，所以我不能在彼此之间嵌套可观察到的数据的情况下，做一个不可靠的解决方案。

我使用以下代码：

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！”一次完成所有请求，而不是为每个可观察到的个人。

编辑：

从我的原始代码中删除undefined将使它工作，但是还有另一个问题导致只执行第一个可以观察到的。

对于远程请求，我使用了requests的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))：

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需要可观测值才能完成，直到转移到下一次，但是对于并发订阅设置为1的mergeMap，情况也是如此。所以那应该没问题。像http请求之类的东西是可以的，因为它们在一次发射后自然完成。websockets或主题或事件发射器的行为会有所不同，必须手动完成，无论是与first或take之类的操作符，还是在源端。

Answer 2

如果你不关心执行的顺序而只想要'Huzzah!‘一旦执行了所有可观察到的操作，要打印，forkJoin也可以是used.Try。

forkJoin(...observables).subscribe(res => console.log('Huzzah');