结合 Observables - 等待所有 observables 触发答案

【问题标题】：Combining Observables - waiting till all observables fire结合 Observables - 等待所有 observables 触发
【发布时间】：2017-05-18 01:53:53
【问题描述】：

我有这个代码：

let $obs = Rx.Observable.merge(
    this.obsEnqueue,
    Rx.Observable.timer(2000)
);

但这不是我想要的——我想要做的是等到两个事件触发（来自this.obsEnqueue 的事件和来自Rx.Observable.timer() 的事件）。

merge 很可能会创建一个新的 Observable，它会在所有传入的 observable 发生第一个事件时触发。我想创建一个新的 observable，它会在所有传入的 observable 都触发后触发。

所以更一般地说：

let $obs = Rx.Observable.X(
    Rx.Observable,    // wait
    Rx.Observable,    // until
    Rx.Observable,    // all of these
    ...
    Rx.Observable     // fire the next event
);

我该怎么做？ X是什么？

【问题讨论】：

This question 是相同的，除了 RX 的 Java 版本。也许你可以为 RXJS 调整答案。
@cdhowie 这个问题略有不同——它说等到所有人完成......这与等到所有人都触发了他们的下一个事件不同吗？对吗？

标签： javascript node.js rxjs5

【解决方案1】：

您可以使用zip，如果您想在所有结果到达时合并它们：

const { Observable } = Rx;

const result = Observable.zip(
  Observable.of('data1'),
  Observable.of(true).delay(500),
  Observable.timer(1000)
);

result.forEach(console.log); // after 1000ms: ['data1', true, 0]

&lt;script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.0.1/Rx.min.js"&gt;&lt;/script&gt;

【讨论】：

zip 在您希望通过等待每个序列每次都提供一个值来组合序列时很有用。另一方面，merge 不会等待，只会发送任何序列中最先出现的值。
谢谢，在 OP 的 cmets 中，有一个与 RxJava 相关的问题，该问题描述了如何等待所有可观察对象完成，这似乎不同，因为 zip 只是等到所有可观察对象都触发了他们的下一个事件（？ ) 但是，该问题中接受的答案也是 zip，这令人困惑。你能解释一下吗？
这里是 Java 问题-stackoverflow.com/questions/35357919/…
zip 可以用于任何一个。 zip 的工作方式是等待其所有可观察对象提供一个值，然后触发其 onNext。如果有任何 observables 传递到 zip 触发 onCompleted，那么 zip 也会这样做。与onError 相同。您可以将zip 视为通过从顶部获取每个元素来合并多个数组。如果我们到达任何数组的末尾，我们就完成了。
好吧，你不能用 zip 真正做到这一点，这不是它的预期用例，除非你给它只调用 onCompleted 而不调用 onNext 的 observables。如果您查找执行此操作的方法，我相信网上有很多示例 (here's one)。此外，您可以随时 implement your own operator 获取您可能需要的一些自定义行为。

【解决方案2】：

你也许可以使用forkJoin:

Rx.Observable.forkJoin(
  Rx.Observable.of('hello').delay(2000),
  Rx.Observable.of('world').delay(1000)
)
  .subscribe(console.log)

两秒钟后，您应该有一个 ['hello', 'world'] 数组。输出的顺序将与参数的输入顺序相同。

输入可以是多个参数、数组和其他东西：

从 Array、类数组对象、Promise、可迭代对象或类 Observable 对象创建 Observable。

这是一个Fiddle 来测试它（只需打开控制台）。

【讨论】：

感谢我尝试了 forkJoin，但它对我不起作用，我会再试一次