累积来自不同 Streams 的 Observables 进行轮询答案

【问题标题】：Accumulate Observables from different Streams for polling累积来自不同 Streams 的 Observables 进行轮询
【发布时间】：2014-03-24 18:54:27
【问题描述】：

我正在尝试从服务器调用中积累 Observables，然后将它们平面映射并进行另一个服务器调用。

private Observable poll(Observable<TaskStatus> taskObservable) {
     Observable.add(taskObservable) //Pseudocode
    .buffer(3 sec) 
    .flatMap(...)
    ...
}

Observable 的累加（“add”）如何实现？

【问题讨论】：

你能举一个期望的输入和输出的例子吗？

标签： java system.reactive reactive-programming rx-java

【解决方案1】：

您可能正在寻找 merge() 运算符。

有关组合 Observables 的更多信息，请参阅：https://github.com/Netflix/RxJava/wiki/Combining-Observables

【讨论】：

如果你同时有两个（或更多）现有的 Observables，合并很棒。Observable.merge(observable1, observable2);这就像调用 ListUtils.union(list1, list2);但我想将它们一一添加到“Empty Observable”中。你明白我的意思吗？
我添加了另一个关于复合一次性/订阅的答案。希望这就是您要寻找的东西？

【解决方案2】：

那么，为了积累，我想你可能会追求像 Rx.NET 中的 CompositeDisposable 这样的东西。

Rx.NET 使用示例：http://rxwiki.wikidot.com/disposables#toc2

Rx.NET 类的文档：http://msdn.microsoft.com/en-us/library/system.reactive.disposables.compositedisposable(v=vs.103).aspx

我相信 Rx Java 等价物是 CompositeSubscription 类。

请看这里：http://netflix.github.io/RxJava/javadoc/rx/subscriptions/CompositeSubscription.html

【讨论】：

【解决方案3】：

您应该使用window 运算符（查看更多信息here，使用window(source, timespan, unit) 搜索版本）。

你的代码应该是这样的：

Observable.window(3, TimeUnit.SECONDS).flatmap(...)

【讨论】：