reactive-programming - Scheduler.elastic() 只创建两个线程，它会在什么基础上创建线程，VM 的 CPU 处理器？

Question

我有一个有 2 个 CPU 的虚拟机。

这是 VM 的 CPU 信息： CPU(s): 2 ; 在线 CPU(s) 列表： 0,1；每个内核的线程数： 1

在其上运行 webflux （在 elastic() 调度程序上运行 1000 个对象的 ParallelFlux）应用程序。它总是只创建 2 个线程来执行不断变化的对象。

尝试使用FromSchedular(Executors.newFixedThreadPool(500))，仍然只创建 2 个线程。

我需要增加 VM 上的 CPU 还是有其他解决方案？

public Flux<FieldResult> execute(List<Field> fieldList){

        return Flux.fromIterable(fieldList)
                   .parallel()
                   .runOn(elastic())
                   .flatMap(field -> {
                      return processField(field);
                   });
}

标签： reactive-programmingspring-webfluxproject-reactor

Answer 1

Flux.parallel ()默认创建与 CPU 内核一样多的轨。

要使用更多导轨，您需要使用Flux.parallel(int parallelism)。之后，您可以使用ParallelFlux.runOn(Schedulers.elastic()).

在您的示例中：

public ParallelFlux<FieldResult> execute(List<Field> fieldList){

    return Flux.fromIterable(fieldList)
               .parallel(fieldList.size())
               .runOn(Schedulers.elastic())
               .map(field -> {
                  return processField(field);
               });
}