java - 限制 Publisher 中的预取

Question

我想创建一个 Flux，它具有在有限预取的情况下按需生成的元素。

我尝试了以下操作，但看起来此代码无法处理背压，因为 generateElements 变得非常大(1011):

AtomicInteger generateElements = new AtomicInteger(0);
Flux<Integer> source = Flux
    .create(emitter -> {
        while (true)
            emitter.next(generateElements.getAndIncrement());
        })
    .subsribeOn(Schedulers.elastic())
    .limitRate(1);

source.take(4).subsribe(System.out::println);
assertThat(generateElements.get()).isEqualTo(5);

如何将我的 Flux 限制为仅一次预取？

Answer 1

您可以使用Flux.generate它需要一个可调用的地方，您可以在其中定义单个项目发射:

AtomicInteger generateElements = new AtomicInteger(0);

Flux.generate(emitter -> emitter.next(generateElements.getAndIncrement()))
    .subscribeOn(Schedulers.elastic())
    .take(4)
    .subscribe(System.out::println);