限制 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在定义单个项目发射的地方使用 which expects a callable：

AtomicInteger generateElements = new AtomicInteger(0);

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