必要时使用 RxJava 背压缓冲区答案

【问题标题】：RxJava Backpressure Buffer when Necessary必要时使用 RxJava 背压缓冲区
【发布时间】：2018-12-06 19:16:52
【问题描述】：

我正在尝试使用 RxJava 获得特定的期望行为，但我不知道该怎么做。

我想要可变长度缓冲区 BackPressureStrategy。

I.E.当没有背压时，我的订阅者会单独发送所有元素。当有背压时，我想获取上次调用我的消费者到现在之间发布的所有新元素（作为列表）。

这可能吗>

【问题讨论】：

【解决方案1】：

您可以使用buffer() 运算符来创建项目列表。运算符的变体允许您设置收集项目的最大大小或最长时间，或使用第二个可观察对象来触发缓冲区结束。

作为一个极端的例子，你可以创建一个流控制 observable 来控制缓冲区。例如，使用debounce() 运算符来控制何时发出缓冲区。通读RxJava examples

【讨论】：

我已经阅读了 RxJava 示例，这就是为什么默认的 buffer() 和 debounce() 操作对我不起作用的原因： 1. buffer()。我希望我的函数尽快被调用。我可以通过非常短的缓冲区最大持续时间来做到这一点。如果这样做，缓冲区实际上并没有帮助我，因为（如果我得到备份）我将有很多待处理的缓冲区，而不是一个大缓冲区。 2. 去抖动（）。这会过滤掉快速事件。我需要捕获所有事件。
在示例代码中，为了使用 buffer() 运算符和 debounce() 作为触发器，debounce() 运算符不会过滤掉单个事件。相反，debounce() 将在间隔中的第一个事件到达时触发，从而导致 buffer() 运算符立即发出。如果在去抖动间隔期间有更多事件到达，debounce() 将不会触发，从而允许buffer() 累积项目。