扫二维码与项目经理沟通
我们在微信上24小时期待你的声音
解答本文疑问/技术咨询/运营咨询/技术建议/互联网交流
Reactor中Thread和Scheduler的区别是什么,相信很多没有经验的人对此束手无策,为此本文总结了问题出现的原因和解决方法,通过这篇文章希望你能解决这个问题。
成都地区优秀IDC服务器托管提供商(成都创新互联公司).为客户提供专业的成都服务器托管,四川各地服务器托管,成都服务器托管、多线服务器托管.托管咨询专线:18982081108
先看一下之前举的Flux的创建的例子:
Fluxflux = Flux.generate( () -> 0, (state, sink) -> { sink.next("3 x " + state + " = " + 3*state); if (state == 10) sink.complete(); return state + 1; }); flux.subscribe(System.out::println);
可以看到,不管是Flux generator还是subscriber,他们实际上都是运行在同一个线程中的。
如果我们想让subscribe发生在一个新的线程中,我们需要新启动一个线程,然后在线程内部进行subscribe操作。
Monomono = Mono.just("hello "); Thread t = new Thread(() -> mono .map(msg -> msg + "thread ") .subscribe(v -> System.out.println(v + Thread.currentThread().getName()) ) ); t.start(); t.join();
上面的例子中,Mono在主线程中创建,而subscribe发生在新启动的Thread中。
很多情况下,我们的publisher是需要定时去调用一些方法,来产生元素的。Reactor提供了一个新的Schedule类来负责定时任务的生成和管理。
Scheduler是一个接口:
public interface Scheduler extends Disposable
它定义了一些定时器中必须要实现的方法:
比如立即执行的:
Disposable schedule(Runnable task);
延时执行的:
default Disposable schedule(Runnable task, long delay, TimeUnit unit)
和定期执行的:
default Disposable schedulePeriodically(Runnable task, long initialDelay, long period, TimeUnit unit)
Schedule有一个工具类叫做Schedules,它提供了多个创建Scheduler的方法,它的本质就是对ExecutorService和ScheduledExecutorService进行封装,将其做为Supplier来创建Schedule。
简单点看Schedule就是对ExecutorService的封装。
Schedulers工具类提供了很多个有用的工具类,我们来详细介绍一下:
Schedulers.immediate():
提交的Runnable将会立马在当前线程执行。
Schedulers.single():
使用同一个线程来执行所有的任务。
Schedulers.boundedElastic():
创建一个可重用的线程池,如果线程池中的线程在长时间内都没有被使用,那么将会被回收。boundedElastic会有一个最大的线程个数,一般来说是CPU cores x 10。 如果目前没有可用的worker线程,提交的任务将会被放入队列等待。
Schedulers.parallel():
创建固定个数的工作线程,个数和CPU的核数相关。
Schedulers.fromExecutorService(ExecutorService):
从一个现有的线程池创建Scheduler。
Schedulers.newXXX:
Schedulers提供了很多new开头的方法,来创建各种各样的Scheduler。
我们看一个Schedulers的具体应用,我们可以指定特定的Scheduler来产生元素:
Flux.interval(Duration.ofMillis(300), Schedulers.newSingle("test"))
publishOn和subscribeOn主要用来进行切换Scheduler的执行上下文。
先讲一个结论,就是在链式调用中,publishOn可以切换Scheduler,但是subscribeOn并不会起作用。
这是因为真正的publish-subscribe关系只有在subscriber开始subscribe的时候才建立。
下面我们来具体看一下这两个方法的使用情况:
publishOn可以在链式调用的过程中,进行publish的切换:
[@Test](https://my.oschina.net/azibug) public void usePublishOn() throws InterruptedException { Scheduler s = Schedulers.newParallel("parallel-scheduler", 4); final Fluxflux = Flux .range(1, 2) .map(i -> 10 + i + ":"+ Thread.currentThread()) .publishOn(s) .map(i -> "value " + i+":"+ Thread.currentThread()); new Thread(() -> flux.subscribe(System.out::println),"ThreadA").start(); System.out.println(Thread.currentThread()); Thread.sleep(5000); }
上面我们创建了一个名字为parallel-scheduler的scheduler。
然后创建了一个Flux,Flux先做了一个map操作,然后切换执行上下文到parallel-scheduler,最后右执行了一次map操作。
最后,我们采用一个新的线程来进行subscribe的输出。
先看下输出结果:
Thread[main,5,main] value 11:Thread[ThreadA,5,main]:Thread[parallel-scheduler-1,5,main] value 12:Thread[ThreadA,5,main]:Thread[parallel-scheduler-1,5,main]
可以看到,主线程的名字是Thread。Subscriber线程的名字是ThreadA。
那么在publishOn之前,map使用的线程就是ThreadA。 而在publishOn之后,map使用的线程就切换到了parallel-scheduler线程池。
subscribeOn是用来切换Subscriber的执行上下文,不管subscribeOn出现在调用链的哪个部分,最终都会应用到整个调用链上。
我们看一个例子:
[@Test](https://my.oschina.net/azibug) public void useSubscribeOn() throws InterruptedException { Scheduler s = Schedulers.newParallel("parallel-scheduler", 4); final Fluxflux = Flux .range(1, 2) .map(i -> 10 + i + ":" + Thread.currentThread()) .subscribeOn(s) .map(i -> "value " + i + ":"+ Thread.currentThread()); new Thread(() -> flux.subscribe(System.out::println), "ThreadA").start(); Thread.sleep(5000); }
同样的,上面的例子中,我们使用了两个map,然后在两个map中使用了一个subscribeOn用来切换subscribe执行上下文。
看下输出结果:
value 11:Thread[parallel-scheduler-1,5,main]:Thread[parallel-scheduler-1,5,main] value 12:Thread[parallel-scheduler-1,5,main]:Thread[parallel-scheduler-1,5,main]
看完上述内容,你们掌握Reactor中Thread和Scheduler的区别是什么的方法了吗?如果还想学到更多技能或想了解更多相关内容,欢迎关注创新互联行业资讯频道,感谢各位的阅读!
我们在微信上24小时期待你的声音
解答本文疑问/技术咨询/运营咨询/技术建议/互联网交流