RXJS 等待数组中的所有 observables 完成(或错误) [英] RXJS Wait for all observables in an array to complete (or error)

问题描述

我正在将 observables 推入这样的数组中......

I'm pushing observables into an array like such...

var tasks$ = [];
tasks$.push(Observable.timer(1000));
tasks$.push(Observable.timer(3000));
tasks$.push(Observable.timer(10000));

我想要一个 Observable，它在所有任务 $ 完成时发出.请记住，在实践中，tasks$ 没有已知数量的 Observable.

I want an Observable that emits when all tasks$ have completed. Keep in mind, in practice, tasks$ doesn't have a known number of Observables.

我已经尝试过 Observable.zip(tasks$).subscribe() 但这似乎在只有 1 个任务的情况下失败，并且让我相信 ZIP 需要一个偶数个元素才能按我期望的方式工作.

I've tried Observable.zip(tasks$).subscribe() but this seems to fail in the event that there is only 1 task, and is leading me to believe that ZIP requires an even number of elements in order to work the way I would expect.

我试过 Observable.concat(tasks$).subscribe() 但 concat 运算符的结果似乎只是一个可观察的数组......例如与输入基本相同.你甚至不能打电话订阅它.

I've tried Observable.concat(tasks$).subscribe() but the result of the concat operator just seems to be an array of observables... e.g. basically the same as the input. You can't even call subscribe on it.

在 C# 中，这类似于 Task.WhenAll().在 ES6 promise 中，它类似于 Promise.all().

In C# this would be akin to Task.WhenAll(). In ES6 promise it would be akin to Promise.all().

我遇到了许多 SO 问题，但它们似乎都在处理等待已知数量的流(例如将它们映射在一起).

I've come across a number of SO questions but they all seem to deal with waiting on a known number of streams (e.g. mapping them together).

推荐答案

如果您想编写一个在所有源 observable 完成时发出的 observable，您可以使用 forkJoin:

If you want to compose an observable that emits when all of the source observables complete, you can use forkJoin:

import { Observable } from 'rxjs/Observable';
import 'rxjs/add/observable/forkJoin';
import 'rxjs/add/operator/first';

var tasks$ = [];
tasks$.push(Observable.timer(1000).first());
tasks$.push(Observable.timer(3000).first());
tasks$.push(Observable.timer(10000).first());
Observable.forkJoin(...tasks$).subscribe(results => { console.log(results); });

这篇关于RXJS 等待数组中的所有 observables 完成(或错误)的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！