java - 如何在 Java Reactor 中设置完全背压驱动的通量？

Question

我的情况是我需要多个工作人员(比如说 2 个)。工作人员必须执行消耗上游事件的任务。

手头的任务消耗一个事件列表，并且具有独立于列表大小的恒定时间。

因此，我希望上游在请求时只提供一个包含所有缓冲事件的列表，一次一个列表。

遗憾的是，大多数方法都实现了预取。发生的事情是即使使用limitRate(1, 0) 上游收到一个太多的onRequest(1)，只是为了补充下游缓冲区。

所以我很难在工作人员可用时生成缓冲列表:它们通常是提前生成的，这与我最大化缓冲列表大小的目标不符。

我如何实现这样的设置？

有没有办法完全禁用预取？

Answer 1

不确定我是否正确理解了问题。显示您当前正在做什么的示例代码会有所帮助。

在 onRequest 之前不从源中提取数据的一种方法是延迟 Flux 的实例化。所以你的代码看起来像这样:

Flux source = Flux.defer(() -> getFluxForUpstreamSource());

另一种使用背压从源中消费的方法是使用 Flux.generate。您的代码看起来像这样:

Flux source = Flux.generate(
        UpstreamSource::getConnection,
        (connection, sink) -> {
            try {
                sink.next(connection.getNext());
            } catch (UpstreamException e) {
                sink.error(e);
            }
            return connection;
        }
);