java - 累积来自不同 Streams 的 Observables 进行轮询

Question

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

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

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

Answer 1

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

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

Answer 2

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

你的代码应该是这样的：

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

Answer 3

对于积累，我想你可能会追求像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