序

本文主要研究一下reactive streams的schedulers

背景

默认情况下Mono以及Flux都在主线程上运行，有时候可能会阻塞主线程，可以通过设定schedulers让其在其他线程运行。

原始输出

没有使用publishOn及subscribeOn时输出如下

11:26:10.668 [main] DEBUG reactor.util.Loggers$LoggerFactory - Using Slf4j logging framework
11:26:11.097 [main] INFO com.example.demo.SchedulerTest - defer thread:[main]
11:26:11.116 [main] INFO com.example.demo.SchedulerTest - filter thread:[main]
11:26:11.116 [main] INFO com.example.demo.SchedulerTest - filter thread:[main]
11:26:11.116 [main] INFO com.example.demo.SchedulerTest - subscribe thread:[main],data :2
11:26:11.116 [main] INFO com.example.demo.SchedulerTest - filter thread:[main]
11:26:11.116 [main] INFO com.example.demo.SchedulerTest - filter thread:[main]
11:26:11.117 [main] INFO com.example.demo.SchedulerTest - subscribe thread:[main],data :4
复制代码

publishOn(给subscriber配置线程)

   @Testpublic void testPublisherThread(){Scheduler pubScheduler = Schedulers.newSingle("pub-thread");Flux.defer(() -> {LOGGER.info("defer thread:[{}]",Thread.currentThread().getName());return Flux.range(1,4);}).filter(e -> {LOGGER.info("filter thread:[{}]",Thread.currentThread().getName());return e % 2 == 0;}).publishOn(pubScheduler).subscribe(e -> {LOGGER.info("subscribe thread:[{}],data :{}",Thread.currentThread().getName(),e);});}
复制代码

输出

11:31:23.691 [main] DEBUG reactor.util.Loggers$LoggerFactory - Using Slf4j logging framework
11:31:23.871 [main] INFO com.example.demo.SchedulerTest - defer thread:[main]
11:31:23.880 [main] INFO com.example.demo.SchedulerTest - filter thread:[main]
11:31:23.881 [main] INFO com.example.demo.SchedulerTest - filter thread:[main]
11:31:23.881 [main] INFO com.example.demo.SchedulerTest - filter thread:[main]
11:31:23.881 [main] INFO com.example.demo.SchedulerTest - filter thread:[main]
11:31:23.881 [publisher-thread-1] INFO com.example.demo.SchedulerTest - subscribe thread:[publisher-thread-1],data :2
11:31:23.881 [publisher-thread-1] INFO com.example.demo.SchedulerTest - subscribe thread:[publisher-thread-1],data :4
复制代码

可以发现，配置publishOn，改变了subscribe的运行线程

subscribeOn(给publisher配置线程)

   @Testpublic void testSubscriberThread() throws InterruptedException {Scheduler subScheduler = Schedulers.newSingle("sub-thread");Flux.defer(() -> {LOGGER.info("defer thread:[{}]",Thread.currentThread().getName());return Flux.range(1,4);}).filter(e -> {LOGGER.info("filter thread:[{}]",Thread.currentThread().getName());return e % 2 == 0;}).subscribeOn(subScheduler).subscribe(e -> {LOGGER.info("subscribe thread:[{}],data :{}",Thread.currentThread().getName(),e);});Thread.sleep(10*1000);}
复制代码

输出如下：

11:31:58.294 [main] DEBUG reactor.util.Loggers$LoggerFactory - Using Slf4j logging framework
11:31:58.528 [subscriber-thread-1] INFO com.example.demo.SchedulerTest - defer thread:[subscriber-thread-1]
11:31:58.532 [subscriber-thread-1] INFO com.example.demo.SchedulerTest - filter thread:[subscriber-thread-1]
11:31:58.532 [subscriber-thread-1] INFO com.example.demo.SchedulerTest - filter thread:[subscriber-thread-1]
11:31:58.532 [subscriber-thread-1] INFO com.example.demo.SchedulerTest - subscribe thread:[subscriber-thread-1],data :2
11:31:58.533 [subscriber-thread-1] INFO com.example.demo.SchedulerTest - filter thread:[subscriber-thread-1]
11:31:58.533 [subscriber-thread-1] INFO com.example.demo.SchedulerTest - filter thread:[subscriber-thread-1]
11:31:58.533 [subscriber-thread-1] INFO com.example.demo.SchedulerTest - subscribe thread:[subscriber-thread-1],data :4
复制代码

可以发现，配置了subscribeOn，所有的都在这个线程运行，包括defer、包括filter、包括subscribe

publishOn和subscribeOn

 @Testpublic void testPublisherAndSubscriberThread() throws InterruptedException {Scheduler pubScheduler = Schedulers.newSingle("publisher-thread");Scheduler subScheduler = Schedulers.newSingle("subscriber-thread");Flux.defer(() -> {LOGGER.info("defer thread:[{}]",Thread.currentThread().getName());return Flux.range(1,4);}).filter(e -> {LOGGER.info("filter thread:[{}]",Thread.currentThread().getName());return e % 2 == 0;}).publishOn(pubScheduler).subscribeOn(subScheduler).subscribe(e -> {LOGGER.info("subscribe thread:[{}],data :{}",Thread.currentThread().getName(),e);});Thread.sleep(10*1000);}
复制代码

输出

11:33:00.964 [main] DEBUG reactor.util.Loggers$LoggerFactory - Using Slf4j logging framework
11:33:01.125 [subscriber-thread-1] INFO com.example.demo.SchedulerTest - defer thread:[subscriber-thread-1]
11:33:01.134 [subscriber-thread-1] INFO com.example.demo.SchedulerTest - filter thread:[subscriber-thread-1]
11:33:01.135 [subscriber-thread-1] INFO com.example.demo.SchedulerTest - filter thread:[subscriber-thread-1]
11:33:01.135 [subscriber-thread-1] INFO com.example.demo.SchedulerTest - filter thread:[subscriber-thread-1]
11:33:01.135 [subscriber-thread-1] INFO com.example.demo.SchedulerTest - filter thread:[subscriber-thread-1]
11:33:01.135 [publisher-thread-2] INFO com.example.demo.SchedulerTest - subscribe thread:[publisher-thread-2],data :2
11:33:01.135 [publisher-thread-2] INFO com.example.demo.SchedulerTest - subscribe thread:[publisher-thread-2],data :4
复制代码

都配置了话，可以看到subscriber运行在publishOn配置的线程，而defer、filter等运行在subscribeOn配置的线程

publishOn及filter

  @Testpublic void testFilterThread(){Scheduler pubScheduler = Schedulers.newSingle("publisher-thread");Flux.defer(() -> {LOGGER.info("defer thread:[{}]",Thread.currentThread().getName());return Flux.range(1,4);}).publishOn(pubScheduler) //NOTE 注意这里放到了filter之前.filter(e -> {LOGGER.info("filter thread:[{}]",Thread.currentThread().getName());return e % 2 == 0;}).subscribe(e -> {LOGGER.info("subscribe thread:[{}],data :{}",Thread.currentThread().getName(),e);});}
复制代码

输出

13:19:01.606 [main] DEBUG reactor.util.Loggers$LoggerFactory - Using Slf4j logging framework
13:19:01.754 [main] INFO com.example.demo.SchedulerTest - defer thread:[main]
13:19:01.766 [publisher-thread-1] INFO com.example.demo.SchedulerTest - filter thread:[publisher-thread-1]
13:19:01.766 [publisher-thread-1] INFO com.example.demo.SchedulerTest - filter thread:[publisher-thread-1]
13:19:01.766 [publisher-thread-1] INFO com.example.demo.SchedulerTest - subscribe thread:[publisher-thread-1],data :2
13:19:01.766 [publisher-thread-1] INFO com.example.demo.SchedulerTest - filter thread:[publisher-thread-1]
13:19:01.766 [publisher-thread-1] INFO com.example.demo.SchedulerTest - filter thread:[publisher-thread-1]
13:19:01.767 [publisher-thread-1] INFO com.example.demo.SchedulerTest - subscribe thread:[publisher-thread-1],data :4
复制代码

这里将publishOn放在了filter之前，可以发现filter线程也变成publisher线程了 在publishOn之后的filter或map等将使用publishOn配置的线程；之前的话，使用的是main线程或subscribeOn配置的线程

subscribeOn及filter

将subscribeOn放在filter之前，跟之后没有区别，因为没有配置publishOn时，subscribeOn作用于所有，包括filter

window scheduler

还可以给window方法设定线程池

   @Testpublic void testWindowScheduler() throws InterruptedException {Scheduler windowScheduler = Schedulers.newSingle("window-thread");Flux.defer(() -> {LOGGER.info("defer thread:[{}]",Thread.currentThread().getName());return Flux.range(1,4);}).delayElements(Duration.ofMillis(200)) //默认会创建parallel线程,作用于subscribe线程.windowTimeout(1, Duration.ofMillis(100), windowScheduler).onErrorReturn(Flux.<Integer>just(-1)).flatMap(e -> {return e.map(item -> item*10);}).subscribe(e -> {LOGGER.info("subscribe thread:[{}],data :{}",Thread.currentThread().getName(),e);});Thread.sleep(10*1000);}
复制代码

输出

14:15:28.523 [main] DEBUG reactor.util.Loggers$LoggerFactory - Using Slf4j logging framework
14:15:28.701 [main] INFO com.example.demo.SchedulerTest - defer thread:[main]
14:15:28.961 [parallel-1] INFO com.example.demo.SchedulerTest - subscribe thread:[parallel-1],data :10
14:15:29.167 [window-thread-1] INFO com.example.demo.SchedulerTest - subscribe thread:[window-thread-1],data :20
14:15:29.370 [window-thread-1] INFO com.example.demo.SchedulerTest - subscribe thread:[window-thread-1],data :30
14:15:29.573 [parallel-4] INFO com.example.demo.SchedulerTest - subscribe thread:[parallel-4],data :40
复制代码

注意delayElements方法默认给subscriber创建了parallel线程 timeout()，skip()等方法也默认会创建线程

scheduleGroup

前面在publishOn以及subscribeOn使用的都是Schedulers.newSingle，也可以使用多个线程组成的group，比如

Scheduler parallelGroup = Schedulers.newParallel("parallel-group", 8);
复制代码

也可以使用elastic类型，比较适合IO类型的操作

   /*** {@link Scheduler} that dynamically creates ExecutorService-based Workers and caches* the thread pools, reusing them once the Workers have been shut down.* <p>* The maximum number of created thread pools is unbounded.* <p>* The default time-to-live for unused thread pools is 60 seconds, use the appropriate* factory to push a different value.* <p>* This scheduler is not restartable.** @param name Thread prefix** @return a new {@link Scheduler} that hosts a fixed pool of single-threaded* ExecutorService-based workers and is suited for parallel work*/public static Scheduler newElastic(String name) {return newElastic(name, ElasticScheduler.DEFAULT_TTL_SECONDS);}
复制代码

实例

  @Testpublic void testElasticGroup() throws InterruptedException {Scheduler elastic = Schedulers.newElastic("elastic-group");Flux.defer(() -> {LOGGER.info("defer thread:[{}]",Thread.currentThread().getName());return Flux.range(1,4);}).filter(e -> {LOGGER.info("filter thread:[{}]",Thread.currentThread().getName());return e % 2 == 0;}).publishOn(elastic).map(e -> {LOGGER.info("map thread:[{}]",Thread.currentThread().getName());return e * 10;}).subscribeOn(elastic).subscribe(e -> {LOGGER.info("subscribe thread:[{}],data :{}",Thread.currentThread().getName(),e);});Thread.sleep(10*1000);}
复制代码

输出

13:58:37.356 [main] DEBUG reactor.util.Loggers$LoggerFactory - Using Slf4j logging framework
13:58:37.514 [elastic-group-2] INFO com.example.demo.SchedulerTest - defer thread:[elastic-group-2]
13:58:37.520 [elastic-group-2] INFO com.example.demo.SchedulerTest - filter thread:[elastic-group-2]
13:58:37.520 [elastic-group-2] INFO com.example.demo.SchedulerTest - filter thread:[elastic-group-2]
13:58:37.520 [elastic-group-2] INFO com.example.demo.SchedulerTest - filter thread:[elastic-group-2]
13:58:37.520 [elastic-group-2] INFO com.example.demo.SchedulerTest - filter thread:[elastic-group-2]
13:58:37.520 [elastic-group-3] INFO com.example.demo.SchedulerTest - map thread:[elastic-group-3]
13:58:37.520 [elastic-group-3] INFO com.example.demo.SchedulerTest - subscribe thread:[elastic-group-3],data :20
13:58:37.521 [elastic-group-3] INFO com.example.demo.SchedulerTest - map thread:[elastic-group-3]
13:58:37.521 [elastic-group-3] INFO com.example.demo.SchedulerTest - subscribe thread:[elastic-group-3],data :40
复制代码

小结

• 命名 这个publishOn及subscribeOn方法名有点晦涩，更直白一点相当于subscriberThreadPools以及publisherThreadPools。
• publishOn与operations的位置

在publishOn之后的filter或map等将使用publishOn配置的线程；之前的话，使用的是main线程或subscribeOn配置的线程

• subscribeOn

在没有配置publishOn，只配置subscribeOn的话，则作用所有

• 方法内置线程 delayElements(),timeout(),skip()内置会使用额外的线程

doc

• schedulers