京公网安备11010502063487号subscribeOn 和 publishOn 到底有什么区别?写在链的哪里有关系吗?多次调用谁生效?阻塞的老代码该放在哪个调度器?为什么 flatMap 里要加 subscribeOn 才并行?这篇在 Reactor 3.8.7 上把每个阶段的线程名打印出来。
没有任何调度器 订阅发生在:main → 源发出:main → 订阅者收到:main subscribeOn(boundedElastic)(放在链的末尾) 订阅发生在:boundedElastic-1 → 源发出:boundedElastic-1 → map 之后:boundedElastic-1 → 订阅者收到:boundedElastic-1 publishOn(parallel) 放在源与 map 之间 订阅发生在:main → 源发出:main → publishOn 之后:parallel-1 → 订阅者收到:parallel-1 subscribeOn(boundedElastic) + publishOn(parallel) 订阅发生在:boundedElastic-1 → 源发出:boundedElastic-1 → publishOn 之后:parallel-2 → 订阅者收到:parallel-2 subscribeOn(single) 再 subscribeOn(boundedElastic) 订阅发生在:single-1 → 源发出:single-1 → 订阅者收到:single-1 publishOn(single) 再 publishOn(boundedElastic) 订阅发生在:main → 源发出:main → 第一个 publishOn 之后:single-1 → 第二个 publishOn 之后:boundedElastic-1 → 订阅者收到:boundedElastic-1
subscribeOn 影响订阅与源,publishOn 影响它之后的阶段。
Mono.delay(50ms) 之后的下游:[delay 完成后的 doOnNext:parallel-3] 在线程 my-completer 里 complete 一个 CompletableFuture:[fromFuture 完成后的 doOnNext:my-completer]
下游线程由上游来源决定。
阻塞源(300 ms),不加 subscribeOn:subscribe() 调用本身用了 310 ms(卡住了调用线程) 阻塞源(300 ms),subscribeOn(boundedElastic):subscribe() 调用用了 2 ms,整体完成用了 305 ms
subscribeOn 决定谁来被阻塞,不会让阻塞变快。
Flux.range(1, 6).publishOn(parallel).map:[1@parallel-4, 2@parallel-4, 3@parallel-4, 4@parallel-4, 5@parallel-4, 6@parallel-4] flatMap 里 4 个阻塞 300 ms 的内层,内层不加调度器:总耗时 1235 ms,线程分布 [1@main, 2@main, 3@main, 4@main] flatMap 里 4 个阻塞 300 ms 的内层,内层加 subscribeOn(boundedElastic):总耗时 304 ms,线程分布 [1@boundedElastic-1, 2@boundedElastic-2, 3@boundedElastic-3, 4@boundedElastic-4]
flatMap 加内层 subscribeOn 才并行。
availableProcessors = 10;200 个阻塞 20 ms 的任务放到 Schedulers.parallel() 上,一共只用了 10 个线程:[parallel-1, parallel-10, parallel-2, parallel-3, parallel-4, parallel-5, parallel-6, parallel-7, parallel-8, parallel-9] 40 个并发阻塞任务在 boundedElastic 上用了 40 个线程,例如 [boundedElastic-1, boundedElastic-10, boundedElastic-11]
parallel 的线程数等于核数,阻塞操作去 boundedElastic。