如何使用响应式扩展按空闲时间对事件进行分组

Question

我有一个问题，响应式扩展似乎特别适合。我有一个事件源，它在短时间内创建事件，中间有相对较长的空闲时间。我希望将这些事件分组成批，(理想情况下)每个事件爆发将在一个批次中结束。使用RxJava，是否有一个好的方法来做到这一点?Observable.buffer(Observable)或Observable.buffer(Func0)看起来很有希望，但也可以使用Observable.window()或Observable. groupbyuntil()。

Answer 1

下面的代码似乎适用于未绑定缓冲区:

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
    }
}

它发出以下内容:

[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]

Answer 2

您提到的任何操作符都可以工作，这取决于您想要的语义。

如果你想让每个组都是一个List，那么使用buffer: https://github.com/Netflix/RxJava/wiki/Transforming-Observables#buffer

如果您希望每个组是一个序列，则使用window: https://github.com/Netflix/RxJava/wiki/Transforming-Observables#window

它们都非常相似，只是它们的输出不同。窗口允许您在每个组中的每个项发出时处理它们，而缓冲区将等待，直到组中的所有项被收集，然后发出。因此，如果您想一起处理每个组中的所有项，则使用buffer。或者，window可以与scan相结合，以顺序地处理每个窗口中的项，并在它们发出时对它们执行有状态操作。

groupByUntil可能不是你想要的，如果时间是你分组的基础，因为groupBy/groupByUntil是关于键分组的。

我看到的一个问题是，缓冲区/窗口通常有固定的间隔，或者需要你通过另一个可观察对象指定当窗口开始和结束。你似乎想要一些在闲置一段时间后触发的东西，这更像是debounce，但debounce只会给你最后一个值，而不是整个组。

你可以做这些复杂的组合，通过多播流，路由一个通过缓冲区，另一个通过debounce，并使用debounce输出信号窗口的开始/结束点。不过这很棘手，我从来没有试过。

现有的缓冲区/窗口用例是否为您工作，或者您是否需要bufferedDebounce行为?