我正在使用Flux和Mono,并在多线程环境下使用它们,使用提供工作线程的Schedular。
有许多选项可以使用elastic、parallel和newElastic来启动Schedular。
以下是我使用的代码:
System.out.println("------ elastic --------- ");
Flux.range(1, 10)
.map(i -> i / 2)
.publishOn(Schedulers.elastic()).log()
.blockLast();
System.out.println("------ new elastic --------- ");
Flux.range(1, 10)
.map(i -> i / 2).log()
.publishOn(Schedulers.newElastic("my")).log()
.blockLast();
它们俩都有相同的文档:
动态创建基于ExecutorService的Workers并缓存线程池,一旦Workers被关闭便重用它们。
创建的线程池数量没有上限。
未使用的线程池的默认生存时间为60秒,使用适当的工厂来推送不同的值。
此调度程序不可重新启动。
以下是它们俩的日志:
------ elastic ---------
[ INFO] (main) | onSubscribe([Fuseable] FluxPublishOn.PublishOnSubscriber)
[ INFO] (main) | request(unbounded)
[ INFO] (elastic-2) | onNext(0)
[ INFO] (elastic-2) | onNext(1)
[ INFO] (elastic-2) | onNext(1)
[ INFO] (elastic-2) | onNext(2)
[ INFO] (elastic-2) | onNext(2)
[ INFO] (elastic-2) | onNext(3)
[ INFO] (elastic-2) | onNext(3)
[ INFO] (elastic-2) | onNext(4)
[ INFO] (elastic-2) | onNext(4)
[ INFO] (elastic-2) | onNext(5)
[ INFO] (elastic-2) | onComplete()
------ new elastic ---------
[ INFO] (main) | onSubscribe([Fuseable] FluxMapFuseable.MapFuseableSubscriber)
[ INFO] (main) | onSubscribe([Fuseable] FluxPublishOn.PublishOnSubscriber)
[ INFO] (main) | request(unbounded)
[ INFO] (main) | request(256)
[ INFO] (main) | onNext(0)
[ INFO] (main) | onNext(1)
[ INFO] (my-4) | onNext(0)
[ INFO] (main) | onNext(1)
[ INFO] (my-4) | onNext(1)
[ INFO] (main) | onNext(2)
[ INFO] (my-4) | onNext(1)
[ INFO] (my-4) | onNext(2)
[ INFO] (main) | onNext(2)
[ INFO] (main) | onNext(3)
[ INFO] (my-4) | onNext(2)
[ INFO] (main) | onNext(3)
[ INFO] (my-4) | onNext(3)
[ INFO] (my-4) | onNext(3)
[ INFO] (main) | onNext(4)
[ INFO] (my-4) | onNext(4)
[ INFO] (main) | onNext(4)
[ INFO] (main) | onNext(5)
[ INFO] (my-4) | onNext(4)
[ INFO] (main) | onComplete()
[ INFO] (my-4) | onNext(5)
[ INFO] (my-4) | onComplete()
这两者之间有什么区别?
elastic
更好,而在什么情况下使用newElastic
更好? - jabrena