RxJava缓冲区与TimeOrCount（）实现？

Question

使用RxJava，我需要将一个项目流缓冲到3个组中，但如果传入项目之间的间隔超过500ms，则刷新缓冲区。

bufferWithTimeOrCount（）操作符正是我想要的，但它似乎只针对RxJS和Rx实现。NET，我需要使用RxJava来实现这一点。

是否有方法复制bufferWithTimeOrCount（）的行为，并获得我对现有RxJava 1的期望。x运算符？

缓冲区（500，TimeUnit.millises，3）尝试每隔500毫秒发出一个新列表，不考虑自上一项以来的时间。

我一直在尝试使用缓冲区（bufferClosingSelector）（请参阅RxJava缓冲区）。我设置了一个序列，该序列发射的项目之间的延迟越来越大：

Observable<Long> emitter = Observable
    .range(1, 9)
    .flatMap(n -> {
        return Observable.just(n).delay(n * n * 50, TimeUnit.MILLISECONDS); 
    })
    .timeInterval()
    .map(interval -> interval.getIntervalInMilliseconds());

然后，当项目之间的时间超过500ms时，我尝试使用debounce（）刷新缓冲区：

emitter
    .buffer(emitter.debounce(500, TimeUnit.MILLISECONDS))
    .toBlocking()
    .subscribe(i -> System.out.println(i));

这似乎可行，产生如下序列：

[51, 150, 250, 352, 449]
[551]
[650]
[749]
[850]

然后我尝试创建一个计数器来在每三个项目后刷新缓冲区，我认为我可以将它与拆散的可观察对象合并（）：

emitter
    .buffer(emitter
       .scan(1L, (n, x) -> n + 1)   // count the items up from 1
       .filter(n -> ((n % 3) == 0)  // emit every three items
    )
    .toBlocking()
    .subscribe(i -> System.out.println(i));

但结果并不那么成功：

[50]
[152, 248, 350]
[452, 550, 650, 749]
[]

如果我输出时间间隔，那么计数器似乎与可观察的批处理异步运行，因此在触发时间和批处理实际释放时间之间会发生竞争情况：

emitter
    .buffer(emitter
       .doOnNext(i -> System.out.println("i = " + i))
       .scan(1L, (n, x) -> n + 1)   // count the items up from 1
       .filter(n -> ((n % 3) == 0)  // emit every three items
    )
    .toBlocking()
    .subscribe(i -> System.out.println(i));

生产：

i = 63
i = 150
[51, 150]
i = 249
i = 350
i = 451
[249, 351]
i = 550
i = 650
i = 750
[450, 550, 650, 750]
i = 850
[850]

...现在我有点困惑。

那很长，但是TL；DR：有没有办法复制bufferWithTimeOrCount（）的行为，并用现有的RxJava 1获得我想要的东西。x运算符？谢谢

Answer 1

编辑：在再次查看该问题后，我意识到该解决方案将始终为每三个项目发出一次，无论它是否已因去抖动而发出。我无法立即想到一种方法，在去抖动后“重置”计数缓冲区。

问题是，通过使用emitter两次，您可以创建两个具有不确定性行为的独立流，因为涉及调度程序。

关键是结合缓冲区（...）、防抖（...）和发布（...）运算符：

emitter.publish(source -> {
    // Create the buffer here using a shared source
    return source.buffer(Observable.defer(() ->
         // Merge these reasons for closing the buffer
         Observable.merge(
             // Either after 500 ms
             source.debounce(500, TimeUnit.MILLISECONDS),
             // or every 3 items
             source.buffer(3)
         )
    ))
})

我无法测试这个解决方案，但这些评论应该可以解释它背后的想法。可观察到的<代码>。延迟确保负责关闭窗口的观察对象是在原始源之后创建的。