Project Reactor 并行执行

Question

Project Reactor 3.1.5.RELEASE

考虑一下：

Flux.range(0, 10)
    .publishOn(Schedulers.parallel())
    .subscribe(i -> LOG.info(i));

我希望订阅者在多个线程中订阅运行，但它运行只在一个线程中：

2018-03-26 12:30:08.693  INFO 89770 --- [     parallel-1] d.a.Application    : 0
2018-03-26 12:30:08.693  INFO 89770 --- [     parallel-1] d.a.Application    : 1
2018-03-26 12:30:08.693  INFO 89770 --- [     parallel-1] d.a.Application    : 2
2018-03-26 12:30:08.693  INFO 89770 --- [     parallel-1] d.a.Application    : 3
2018-03-26 12:30:08.694  INFO 89770 --- [     parallel-1] d.a.Application    : 4
2018-03-26 12:30:08.694  INFO 89770 --- [     parallel-1] d.a.Application    : 5
2018-03-26 12:30:08.694  INFO 89770 --- [     parallel-1] d.a.Application    : 6
2018-03-26 12:30:08.694  INFO 89770 --- [     parallel-1] d.a.Application    : 7
2018-03-26 12:30:08.694  INFO 89770 --- [     parallel-1] d.a.Application    : 8
2018-03-26 12:30:08.694  INFO 89770 --- [     parallel-1] d.a.Application    : 9

文档告诉我的期望是正确的 (http://projectreactor.io/docs/core/release/reference/#threading)。有人可以向我解释一下那里发生了什么吗？

Answer 1

反应流本质上是顺序的，publishOn 只是告诉源在哪里一个接一个地发出每个值。您需要通过 parallel 告诉流程并行进行，然后通过 runOn:

指定调度程序

Flux.range(0, 10)
.parallel()
.runOn(Schedulers.parallel())
.doOnNext(i -> LOG.info(i))
.sequential()
.subscribe();

Project Reactor 并行执行

Project Reactor parallel execution

java

reactive-programming

project-reactor