聊聊reactive streams的schedulers
序
本文主要研究一下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
聊聊reactive streams的schedulers相关推荐
- 聊聊reactive streams的processors
序 本文主要研究一下reactive streams的processors processors分类 processors既是Publisher也是Subscriber.在project reacto ...
- Reactive Streams
2019独角兽企业重金招聘Python工程师标准>>> Reactive Streams is an initiative to provide a standard for asy ...
- Reactive Streams规范及常见库
一.什么是Reactive Streams Reactive Streams is an initiative to provide a standard for asynchronous strea ...
- JDK11的新特性:HTTP API和reactive streams
文章目录 简介 怎么在java中使用reactive streams POST请求的例子 总结 简介 在JDK11的新特性:新的HTTP API中,我们介绍了通过新的HTTP API,我们可以发送同步 ...
- 什么是Reactive Streams in Java 译
数式编程对于Reactive Programming很重要,但我不会在这篇文章中深入探讨函数式编程. 在这篇文章中,我想看看Java中的整体Reactive发展环境. Reactive Program ...
- JVM平台上的响应式流(Reactive Streams)规范
Reactive Streams 响应式流是一个倡议,用来为具有非阻塞后压的异步流处理提供一个标准.大家努力的目标集中在运行时环境(JVM和JavaScript)和网络协议上. 注:响应式流其实就是一 ...
- Reactive Streams规范
Reactive Streams的目的是提供一个带有 非阻塞背压 特征的 异步流处理 标准. 最新的发布版本如下 <dependency><groupId>org.reacti ...
- reactive streams的Mono及Flux
实例 Mono @Testpublic void testMonoBasic(){Mono.fromSupplier(() -> "Hello").subscribe(Sys ...
- FunDA(6)- Reactive Streams:Play with Iteratees、Enumerator and Enumeratees
在上一节我们介绍了Iteratee.它的功能是消耗从一些数据源推送过来的数据元素,不同的数据消耗方式代表了不同功能的Iteratee.所谓的数据源就是我们这节要讨论的Enumerator.Enumer ...
最新文章
- 六年级计算机word处理,六年级上信息技术教案Word大变身用Word制作网页河大版
- 如何简单形象又有趣地讲解神经网络是什么?(知乎) 说的人很多,理解很充分_kebu12345678的博客-CSDN博客_神经网络知乎
- NSCalendar 日历类
- 4款基于jquery的列表图标动画切换特效
- fpga硬件驱动TFT液晶屏-小结
- LeetCode 1851. 包含每个查询的最小区间(排序 + 离线查询 + 优先队列)
- mysql的Innodb为什么使用B+树
- python牛客网编程题_一波优秀的自学编程语言网站
- android 编译faac,faac 移植到android
- 一张壁纸就是一段故事,壁纸控的找图不难!
- Linux性能优化之内存优化(二)
- java中PrepareStatement使用的一点小问题
- 坑爹的libxml2 for mingw 编译
- 数据结构与算法之队列
- 小马哥经典摘抄语录之 ————-有创业及野心的一定要看
- delphi创建excel写入数据并创建二维折线图
- Plague Inc
- 我被这个浏览了 746000 次的问题惊住了!
- 免费的 ChatGPT镜像网站
- signal,blinker:信号(看我脸色行事)
热门文章
- java invoker_Spring的HTTPInvoker
- 值得关注!一种新型脑机接口--集成光子芯片的脑机接口是否可行?
- Oculus Go 开发者版本开放索取,下个月设备开始配送
- 不拆分单词也可以做NLP,哈工大最新模型在多项任务中打败BERT,还能直接训练中文...
- 复旦教授发现400+安卓漏洞,最严重可使手机变砖,谷“鸽”16个月后才修复
- 丁磊旗下首个IPO的AI业务!网易有道年入7个亿,清华96计算机系为主力,沈向洋任董事...
- NeurIPS审稿引发吐槽大会,落选者把荒唐意见怼了个遍:“我谢谢你们了”
- 新工具一键安装Java环境!微软又双叒叕造福开发者
- 想在PyTorch里训练BERT,请试试Facebook跨语言模型XLM
- mormot支持websocket