java - Flux.generate 与 Consumer 和并行

Question

我有简单的通量

Flux<Long> flux = Flux.generate(
            AtomicLong::new,
            (state, sink) -> {
                long i = state.getAndIncrement();
                sink.next(i);
                if (i == 3) sink.complete();
                return state;
            }, (state) -> System.out.println("state: " + state));

在单个线程中按预期工作:

flux.subscribe(System.out::println);

输出为

0 1 2 3 state: 4

但是当我切换到并行时:

flux.parallel().runOn(Schedulers.elastic()).subscribe(System.out::println);

应打印状态的消费者:未调用 Number。我刚刚看到:

0 3 2 1

这是一个错误还是预期的功能？

Answer 1

我不是响应式(Reactive)专家，但在深入研究源代码后，似乎这种行为是设计使然；看来创建ParallelFlux有阻塞State Consumer调用的副作用；如果您想并行并调用 State Consumer，您可以使用:

flux.publishOn(Schedulers.elastic()).subscribe(System.out::println);