我有一个反应式扩展似乎特别适合的问题。我有一个事件源,它在较短的突发中创建事件,其间有相对较长的空闲期。我想将这些事件分批分组,其中(理想情况下)每次突发事件都会以一批结束。使用 RxJava,有没有好的方法来做到这一点?Observable.buffer(Observable)或Observable.buffer(Func0)看起来很有希望,但使用Observable.window()或 Observable.groupByUntil() 可能是可行的。
2 回答
Here is code that seems to work for a debounced buffer:
import java.util.List;
import java.util.concurrent.TimeUnit;
import rx.Observable;
import rx.Subscriber;
import rx.schedulers.Schedulers;
public class DebounceBuffer {
public static void main(String args[]) {
// see all bursts in a single sequence
// intermittentBursts().toBlocking().forEach(System.out::println);
// debounce to the last value in each burst
// intermittentBursts().debounce(10, TimeUnit.MILLISECONDS).toBlocking().forEach(System.out::println);
/* The following will emit a buffered list as it is debounced */
// first we multicast the stream ... using refCount so it handles the subscribe/unsubscribe
Observable<Integer> burstStream = intermittentBursts().publish().refCount();
// then we get the debounced version
Observable<Integer> debounced = burstStream.debounce(10, TimeUnit.MILLISECONDS);
// then the buffered one that uses the debounced stream to demark window start/stop
Observable<List<Integer>> buffered = burstStream.buffer(debounced);
// then we subscribe to the buffered stream so it does what we want
buffered.take(20).toBlocking().forEach(System.out::println);
}
public static Observable<Integer> intermittentBursts() {
return Observable.create((Subscriber<? super Integer> s) -> {
while (!s.isUnsubscribed()) {
// burst some number of items
for (int i = 0; i < Math.random() * 20; i++) {
s.onNext(i);
}
try {
// sleep for a random amount of time
Thread.sleep((long) (Math.random() * 1000));
} catch (Exception e) {
// do nothing
}
}
}).subscribeOn(Schedulers.newThread()); // use newThread since we are using sleep to block
}
}
It emits the following:
[0, 1, 2, 3, 4, 5, 6, 7]
[0, 1, 2, 3, 4, 5]
[0, 1, 2, 3, 4, 5, 6, 7, 8]
[0, 1]
[0, 1, 2, 3, 4, 5]
[0, 1, 2]
[0, 1, 2, 3, 4, 5, 6, 7]
[0, 1, 2, 3, 4, 5, 6, 7, 8]
[0, 1, 2, 3, 4, 5, 6]
[0, 1, 2]
[0, 1, 2, 3, 4, 5, 6, 7, 8]
[0, 1, 2, 3, 4]
[0, 1, 2, 3]
[0, 1, 2, 3, 4, 5, 6, 7, 8]
[0, 1, 2, 3, 4, 5, 6, 7]
[0, 1, 2, 3, 4, 5, 6, 7, 8]
[0, 1, 2, 3]
[0]
[0, 1, 2]
[0]
您提到的任何这些运算符都可以使用,这取决于您想要什么语义。
如果您希望每个组成为一个列表,请使用缓冲区:https ://github.com/Netflix/RxJava/wiki/Transforming-Observables#buffer
如果您希望每个组都是一个序列,则使用窗口:https ://github.com/Netflix/RxJava/wiki/Transforming-Observables#window
两者非常相似,只是它们的输出不同。Window 允许您在每个组中的每个项目发出时对其进行处理,而缓冲区将等待直到组中的所有项目都被收集然后发出。因此,如果您想一起处理每个组中的所有项目,请使用缓冲区。或者,window 可以与 scan 相结合,在每个窗口中按顺序处理项目,并在它们发出时对其进行有状态操作。
如果时间是您分组的依据,groupByUntil 可能不是您要寻找的,因为 groupBy/groupByUntil 是关于键分组的。
我可以看到缓冲区/窗口的一个问题是它们通常具有固定的间隔,或者需要您在窗口开始和结束时通过另一个 Observable 指定。你似乎想要在一段空闲时间后触发的东西,这更像是 debounce,但 debounce 只会给你最后一个值,而不是整个组。
您可以对这些进行复杂的组合,通过多播流,通过缓冲区路由一个,另一个通过去抖动,并使用去抖动输出来通知窗口开始/结束点。不过这很棘手,我从未尝试过。
现有的缓冲区/窗口用例是否适合您,或者您是否需要 bufferedDebounce 行为?