rxJava buffer() 与时间背压

本文介绍了rxJava buffer() 与时间背压的处理方法，对大家解决问题具有一定的参考价值，需要的朋友们下面随着小编来一起学习吧！ 问题描述

buffer 操作符的版本不按 JavaDoc 进行时间荣誉背压:

The versions of buffer operator that don't operate on time honour backpressure as per JavaDoc:

reactivex.io/RxJava/2.x/javadoc/io/reactivex/Flowable.html#buffer-int-

然而，任何涉及基于时间的缓冲区的buffer版本都不支持背压，就像这个

However, any version of buffer that involves time based buffers doesn't support backpressure, like this one

reactivex.io/RxJava/2.x/javadoc/io/reactivex/Flowable.html#buffer-long-java.util.concurrent.TimeUnit-int-

我理解这是因为一旦时间在流逝，您就无法停止它，例如 interval 运算符，出于同样的原因，它不支持背压.

I understand this comes from the fact that once the time is ticking, you can't stop it similarly to, for example interval operator, that doesn't support backpressure either for the same reason.

我想要的是一个缓冲操作符，它基于大小和时间，并且通过将背压信号传播到上游和时间滴答生成器来完全支持背压，如下所示:

What I want is a buffering operator that is both size and time based and fully supports backpressure by propagating the backpressure signals to BOTH the upstream AND the time ticking producer, something like this:

someFlowable() .buffer( Flowable.interval(1, SECONDS).onBackpressureDrop(), 10 );

所以现在我可以放弃反压信号了.

So now I could drop the tick on backpressure signals.

这是目前在 rxJava2 中可以实现的吗?Project-Reactor 怎么样?

Is this something currently achievable in rxJava2? How about Project-Reactor?

推荐答案

我最近遇到了这个问题，这里是我的实现.可以这样使用:

I've encountered the problem recently and here is my implementation. It can be used like this:

Flowable<List<T>> bufferedFlow = (some flowable of T) pose(new BufferTransformer(1, TimeUnit.MILLISECONDS, 8))

它支持您指定的数量的背压.

It supports backpressure by the count you've specified.

以下是实现:gist.github/driventokill/c49f86fb0cc1829a2a2e>

Here is the implementation: gist.github/driventokill/c49f86fb0cc182994ef423a70e793a2d