专栏首页码匠的流水账聊聊reactive streams的schedulers

聊聊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配置线程)

    @Test
    public 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配置线程)

    @Test
    public 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

    @Test
    public 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

    @Test
    public 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方法设定线程池

    @Test
    public 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);
    }

实例

    @Test
    public 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

本文分享自微信公众号 - 码匠的流水账(geek_luandun)

原文出处及转载信息见文内详细说明,如有侵权,请联系 yunjia_community@tencent.com 删除。

原始发表时间:2018-01-15

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • 解决jd-gui在Sierra下闪退问题

    在升级了mac操作系统到Sierra版本之后,之前的jd-gui就闪退了,本文就讲述一下如何解决这个问题。

    codecraft
  • 关于ribbonClient配置的一个坑

    不知道从哪个版本起,给ribbon配置物理的server list起,单纯配置xxx.ribbon.listOfServers不起效果了,于是就开启了埋坑之旅。

    codecraft
  • 聊聊dkron的Scheduler

    dkron的Scheduler定义了Cron、Started、EntryJobMap属性;NewScheduler方法创建默认的Scheduler;它提供了St...

    codecraft
  • Servlet 3.0 异步处理详解

    Servlet 3.0 开始提供了AsyncContext用来支持异步处理请求,那么异步处理请求到底能够带来哪些好处?

    颇忒脱
  • Innodb存储引擎中的后台线程介绍

    在Innodb存储引擎中,后台线程的主要作用是负责刷新内存池中的数据,保证缓冲池中的内存缓存的是最近的数据。此外它会将已经修改的数据文件刷新到磁盘文件中...

    AsiaYe
  • gevent初探

    有很多Python语言的协程库,如:tornado、asyncio等。这些库在使用时需要使用特定的语法,如:async/await,对于非协程的代码需要改写才能...

    drunkdream
  • MySQL实例阻塞分析一例(线程statistics状态)

    某日下午下班后低峰期,现网MySQL一个库突然报出大量慢sql,状态是 statistics,但是过后拿这些sql去执行的时候,实际很快。处于 statisti...

    wubx
  • MySQL实例阻塞分析一例(线程statistics状态)

    某日下午下班后低峰期,现网MySQL一个库突然报出大量慢sql,状态是 statistics,但是过后拿这些sql去执行的时候,实际很快。处于 statisti...

    wubx
  • 解决jd-gui在Sierra下闪退问题

    在升级了mac操作系统到Sierra版本之后,之前的jd-gui就闪退了,本文就讲述一下如何解决这个问题。

    codecraft
  • NumPy快速入门--基础知识

    与许多矩阵语言不同,乘法运算符 * 的运算在NumPy数组中是元素级别的。矩阵乘积可以使用 dot 函数或方法执行:

    Michael阿明

扫码关注云+社区

领取腾讯云代金券