如何使用RxJava实现缓冲区，其大小取决于消费者和生产者的速度？

Question

要求如下：

1. 当生产者比消费者快时，所生产的物品应进行缓冲，直到缓冲满，当缓冲满时，不得再要求生产者生产新物品（背压）。
2. 当消费者比生产者快的时候，缓冲的物品会被释放出来进行补偿。如果缓冲区是空的，那么就会发生饥饿，但是在这种情况下我们无能为力。
3. 缓冲区的目标是降低饥饿的风险；因此，只要生产者比消费者快，它就应该保持满。

我尝试过不同的可流动操作符。最接近这些要求的是具有缓冲区大小的observeOn。但这个操作符背后有一些魔力：当缓冲区满时，它会逐渐清空，直到其大小小于其总容量的25%，在这个过程中，不会向生产商请求任何项目。

我应该使用哪个操作员或操作员组合来满足这些要求？

提前感谢。

Answer 1

我最终得出结论，一个定制的可流动操作员可以是一个可接受的解决方案，既可以管理溢出（通过背压），也可以管理下溢（通过一个缓冲区，它实际上充当元素的储存器，以避免饥饿）。

当该操作员连接到可流动管道时，元素储液罐中充满了生产商排放的元素。消费者根据需要获取储存在储液罐中的元素。生产者和消费者在两个不同的调度器上运行。

溢出情况：

• 当储层充满时，生产商不再被要求排放新的元素（由于管道的背压设施）

下溢情况：

• 储液罐用于根据需要向消费者提供元素

生命周期

• 缓冲阶段：储液罐充满，直到满为止。在此阶段，消费者不会收到任何元素
• 转移阶段：消费者在需要时获取储存在储液罐中的元素；只要储液罐未满，生产者就会将元素储存在储液罐中
• 冲洗阶段：生产商已完成；耗电元件接收储液罐中剩余的所有元件，并最终接收到接通完整信号

该运算符在溢出和下限溢位情况下都是非阻塞的。

用法：

ReservoirOperator op = new ReservoirOperator(bufferSize, Schedulers.io());
upstream.lift(op).subscribe(consumer);

Answer 2

使用多个rebatchRequests（在封面下使用observeOn）人为地提高进一步请求的级别。