Promise.all() with RxJS

Question

我正在用 Angular 2 编写一个应用程序，我想执行几个 http 请求并在响应上运行一个函数。

在 Angular 1 中，我会写类似 $q.all([$http.get(...), $http.get(...), ...]).then(doSomethingWithResponses);

但是 Angular 2 返回了 RxJS Observables，经过一堆阅读，我仍然无法弄清楚如何获得几个 http 请求的响应。 如何做到这一点？

Answer 1

正如@Eric马丁内斯指出的那样，有forkJoin。 forkJoin 并行运行所有可观察序列并收集它们的最后一个元素。

Rx.Observable.forkJoin([a,b]).subscribe(t=> {
        var firstResult = t[0];
        var secondResult = t[1];
});

Answer 2

我不确定您是否要使用forkJoin/zip，特别是考虑到combineLatest更容易理解并且会在每个子流事件上发出，而forkJoin基本上是每个子流上的样本都发出了。

当您想在未来组合多项目可观察量时，这可能会在以后咬您。

Answer 3

merge不

行吗？您可以订阅处理程序并将其附加到onComplete回调。

我首先构建一个可观察数组，然后使用静态merge：

let obs_ary: any = [obs1, obs2, obs3];
Observable.merge(...obs_ary);

Answer 4

我正在学习 RxJS，我试图用 RxJS v5 做同样的事情

似乎我们在 v5 上不再有forkJoin了，所以这是我让它工作的方式（适用于 flatMap 或 mergeMap ，它们是别名）：

const callOne = value =>
    new window.Promise(resolve =>
        setTimeout(() => resolve(value + 10), 3000)
    );
const callTwo = value =>
    new window.Promise(resolve =>
        setTimeout(() => resolve(value + 20), 1000)
    );
Rx.Observable
    .of(2)
    .do(() => console.log('querying...'))
    .mergeMap(number =>
        Rx.Observable.zip(
            Rx.Observable.fromPromise(callOne(number)),
            Rx.Observable.fromPromise(callTwo(number))
        )
    ).concatAll()
    .subscribe(createSubscriber('promises in parallel'));