从 Flux 创建一个惰性流

Question

我正在尝试使用 Californium-SR10 中的 Project Reactor 创建一个 lazy-Stream。

根据the javadoc：

在 Iterator.next() 调用上将此 Flux 转换为惰性 Iterable 阻塞。

因此，我尝试了以下方法：

AtomicInteger generatedElements = new AtomicInteger(0);
Flux<Integer> source = Flux
    .range(0, 10)
    .doOnRequest(req -> System.out.println("Requested " + req))
    .doOnRequest(req -> generatedElements.addAndGet((int) req))
    .limitRate(2)
    .subscribeOn(Schedulers.elastic());

Iterator<Integer> l = source.toIterable().iterator();
assertThat(l.next()).isEqualTo(0);
assertThat(l.next()).isEqualTo(1);
assertThat(l.next()).isEqualTo(2);

Thread.sleep(2000);
assertThat(generatedElements.get()).isEqualTo(4);

这给了我以下令人惊讶的结果：

Requested 2
Requested 2
Requested 2
Requested 2
Requested 2
Requested 2


org.junit.ComparisonFailure: expected:<[4]> but was:<[12]>
Expected :4
Actual   :12

您对这里发生的事情（以及如何解决）有任何解释吗？

Answer 1

为什么每个参数都传给doOnRequest()2？

doOnRequest() 将在订阅者从上游请求新元素时触发（连同请求的​​元素数量作为其参数）。由于您将速率限制为 2，因此您希望使用 2 调用它每次都是一个参数，它正在这样做。

为什么Flux 不能懒惰地完成？

嗯，实际上它确实，但不是你期望的那样。 Lazy 并不一定意味着“一次一个”，它只是意味着它将根据需要评估 批次，而不是总是一次性评估整个 Flux。

具体来说，请注意limitRate() 方法不会以同样的方式自动应用于迭代器 - 它有一个您必须指定的单独批量大小作为toIterable() 方法的参数。

您可以指定预取率和批量大小为 1，然后您可能只生成 4 个元素（而不是 3 个，因为它将始终为以下next() 调用准备好至少一个附加元素）：

AtomicInteger generatedElements = new AtomicInteger(0);
Flux<Integer> source = Flux
        .range(0, 10)
        .limitRate(1)
        .doOnRequest(req -> System.out.println("Requested " + req))
        .doOnRequest(req -> generatedElements.addAndGet((int) req));

Iterator<Integer> l = source.toIterable(1).iterator();
assertThat(l.next()).isEqualTo(0);
assertThat(l.next()).isEqualTo(1);
assertThat(l.next()).isEqualTo(2);

assertThat(generatedElements.get()).isEqualTo(4);

但是，请注意，即使在所有情况下都不能保证这一点，因为它仅取决于何时应用背压，以及 Flux 何时响应它。