聊聊reactive streams的schedulers
序
本文主要研究一下reactive streams的schedulersreact
背景
默認狀況下Mono以及Flux都在主線程上運行，有時候可能會阻塞主線程，能夠經過設定schedulers讓其在其餘線程運行。app
原始輸出
沒有使用publishOn及subscribeOn時輸出以下ui
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);
}
輸出以下：rest
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);
}
輸出code
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);
});
}
輸出ip
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方法設定線程池get
@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);
}
輸出it
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，好比io
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
相關文章
- 1. 聊聊reactive streams的processors
- 2. 聊聊reactive streams的parallel flux
- 3. 聊聊reactive streams的backpressure
- 4. 聊聊reactive streams的Mono及Flux
- 5. 聊聊reactive streams publisher的doOn方法
- 6. 聊聊reactive streams的tranform操做
- 7. Spring Boot Reactive Streams
- 8. Java 9 Reactive Streams
- 9. java9之Reactive Streams
- 10. What can Reactive Streams offer EE4J?
- 更多相關文章...
- • PHP imagecolorclosest - 取得與指定的顏色最接近的顏色的索引值 - PHP參考手冊
- • PHP imagecolorclosesthwb - 取得與指定的顏色最接近的色度的黑白色的索引 - PHP參考手冊
- • 漫談MySQL的鎖機制
- • 互聯網組織的未來：剖析GitHub員工的任性之源
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
最新文章
- 1. 融合阿里雲，牛客助您找到心儀好工作
- 2. 解決jdbc(jdbctemplate)在測試類時不報錯在TomCatb部署後報錯
- 3. 解決PyCharm GoLand IntelliJ 等 JetBrains 系列 IDE無法輸入中文
- 4. vue+ant design中關於圖片請求不顯示的問題。
- 5. insufficient memory && Native memory allocation (malloc) failed
- 6. 解決IDEA用Maven創建的Web工程不能創建Java Class文件的問題
- 7. [已解決] Error: Cannot download ‘https://start.spring.io/starter.zip?
- 8. 在idea讓java文件夾正常使用
- 9. Eclipse啓動提示「subversive connector discovery」
- 10. 帥某-技巧-快速轉帖博主文章（article_content）
歡迎關注本站公眾號,獲取更多信息
相關文章
- 1. 聊聊reactive streams的processors
- 2. 聊聊reactive streams的parallel flux
- 3. 聊聊reactive streams的backpressure
- 4. 聊聊reactive streams的Mono及Flux
- 5. 聊聊reactive streams publisher的doOn方法
- 6. 聊聊reactive streams的tranform操做
- 7. Spring Boot Reactive Streams
- 8. Java 9 Reactive Streams
- 9. java9之Reactive Streams
- 10. What can Reactive Streams offer EE4J?