Rxjava - Group/Batch 可观察序列中的元素突发

Question

我有一个可观察的序列。当插入第一个元素时，我想启动一个计时器并在计时器的时间跨度内批量处理后续插入的元素。然后，在序列中插入另一个元素之前，计时器不会再次启动。

--------|=====timespan====|---------------|=====timespan====|-------------->
        1  2 3 4    5                     6

会产生：

[1,2,3,4,5], [6]

我尝试使用 Observable.buffer() 和 timespan 但是从我的实验中，我可以看到定时器在我们订阅可观察序列后立即启动，并在前一个时重新启动计时器完成。

所以与前面的例子有相同的序列，并使用 buffer() 和 timespan，我会得到这样的东西：

|=====timespan====|=====timespan====|=====timespan====|=====timespan====|-->
        1  2 3 4                          5 6

这会产生这个：

[1,2,3,4], [], [5,6], []

这与本质上是同一个问题，但 java 代替。

所以问题是，因为我不想过多地延迟我的流，所以我想要一个非常短的计时器，而且那个计时器会非常密集。我可以简单地过滤空列表，但我认为这对 CPU 的影响太大。

Answer 1

window 运算符将充当 buffer，您不能直接使用它。

想法是通过第一个可观测值（我称之为 insertions）的发射来控制 timer。为此，您必须在 link 两个可观察对象中包含第三个参数（stopWatch Subject 在下面的解决方案中）。

    @Test
    public void stop_watch_observable() {

        Subject<Long> stopWatch = PublishSubject.create();

        Observable<Long> insertions = insertions();

        //share to use it as a timer (looking for the first emission)
        //and to recieve the items
        Observable<Long> shared = insertions.share();

        //for each emission of insertions we start a new timer
        //but only the first one is emitted
        //the others are stopped by the takeUntil(stopWatch)
        Observable<Long> window = shared
                .flatMap(e -> Observable.timer(3, TimeUnit.SECONDS).takeUntil(stopWatch));

        shared.buffer(window)
                //each time a window is generated we kill all the current timers
                .doOnNext(e -> stopWatch.onNext(0L))
                .blockingSubscribe(System.out::println);
    }

    // insertions generator which is comming randomly
    private Observable<Long> insertions() {
        AtomicLong al = new AtomicLong(0);
        return Observable.generate((Emitter<Long> emitter) -> {
            if (al.getAndIncrement() % 4 == 0) {
                Long timeToWait = Long.parseLong(RandomStringUtils.randomNumeric(1));
                System.out.println("sleeping for: " + timeToWait);
                sleep(timeToWait * 1000);
            } else {
                sleep(500);
            }
            emitter.onNext(al.get());
        }).subscribeOn(Schedulers.newThread());
    }

Answer 2

第一个解决方案的缺点是每次插入的发射都会启动 timer（它可能是 CPU 密集）。这里是另一种只有一个计时器的解决方案在时间开始（我认为这样更有效：

@Test
    public void stop_watch_observable() {

        Observable<Long> insertions = insertions();
        Observable<Long> shared = insertions.share();

        AtomicBoolean timerOn = new AtomicBoolean(false);

        Observable<Long> window = shared
                .flatMap(e -> timerOn.get() ? Observable.empty() : Observable.timer(3, TimeUnit.SECONDS)
                        .doOnSubscribe(x -> timerOn.set(true))
                );

        shared.buffer(window)
                //each time a window is generated we kill all the current timers
                .doOnNext(e -> timerOn.set(false))
                .blockingSubscribe(System.out::println);
    }

Rxjava - Group/Batch 可观察序列中的元素突发

Rxjava - Group/Batch bursts of elements in an observable sequence

java

rx-java

rx-java2