Flux / Mono 自定义线程调度线程调度主要讲Schedulers/publishOn/subscribeOn前置你已经学过handle、generate、create、buffer、limitRate、doOnXXX、onXxx、Sinks。本文聚焦「响应式代码跑在哪些线程上」——这是把响应式程序从「能跑」变成「跑得对、跑得稳」的分水岭。1. 一句话定位Scheduler是什么Reactor 默认在调用线程上同步执行。一旦要「异步」「并行」「把阻塞 IO 隔离」就必须把部分链路交给一个Scheduler。Scheduler 响应式世界里的「线程池抽象」。它不直接暴露Thread而是提供Worker一个可调度任务的执行单元。你通过publishOn(scheduler)/subscribeOn(scheduler)把链路贴到某个调度器上。关键认知Reactor 不会自动帮你切线程。你不调publishOn/subscribeOn一切都还在当前线程通常是main或 Netty EventLoop上跑——在 EventLoop 上 sleep/阻塞会直接拖垮整个服务。2. 内置Schedulers速查表调度器线程模型适合场景备注Schedulers.immediate()当前线程不切调试、想「就在这线程跑」不是新线程Schedulers.single()全局单条 Worker 线程需要严格串行化的任务进程内共享单例Schedulers.parallel()固定NCPU核数条线程CPU 密集型计算默认并行度Runtime.getRuntime().availableProcessors()Schedulers.boundedElastic()弹性线程池有上限阻塞 IODB/HTTP/文件默认上限 10×核数线程 10万任务队列防 OOMSchedulers.fromExecutor(Executor)用你给的线程池接入既有线程池自定义调度器入口Schedulers.fromExecutorService(ExecutorService)同上可关闭接入既有线程池同上Schedulers.newParallel(name, n)新建独立 parallel 池隔离某类计算用dispose()释放Schedulers.newBoundedElastic(...)新建独立弹性池隔离某类 IO用dispose()释放Schedulers.newSingle(name)新建独立单线程池隔离某类串行任务用dispose()释放黄金法则CPU 密集 / 纯计算 →parallel别在它上面阻塞。阻塞 IO / 等待外部 →boundedElastic别在它上面跑重计算浪费线程。不同业务线要隔离线程池避免 A 业务 IO 拥塞把 B 业务拖死→ 自定义调度器。3. 自定义调度器本文重点3.1 基于普通线程池Schedulers.fromExecutorService最常用场景接入项目里已有的ExecutorService或给线程池命名排查问题时一眼看出是哪条业务线。3.2 基于 Reactor 自带的newXxx当你需要独立、可控、用完可释放的池子时用它们而不是污染全局共享池。3.3 命名线程工厂排查利器默认线程名是parallel-1、boundedElastic-3这种看不出业务。自定义ThreadFactory给线程加前缀线上排查线程 dump 时能直接定位。4.publishOnvssubscribeOn位置语义最重要这是全文最容易被搞混的点务必结合第 5 节的样例亲手跑一遍。4.1publishOn(scheduler)—— 改变「它之后」的线程publishOn作用在**它下游代码顺序中它下方**的所有操作符直到遇到下一个publishOn。每遇到一个publishOn后续链路就被切到那个调度器。可多次叠加每加一个就再切一次。4.2subscribeOn(scheduler)—— 决定「订阅与源」发生在哪subscribeOn决定订阅动作本身 上游源的产生在哪个线程上发生。它影响的是「链的起点」对「下游已经被publishOn接管的处理线程」没有影响。多次subscribeOn时最靠近 source上游的那一个生效代码里最靠上的那个。记忆法publishOn从它往下换线程下游视角。subscribeOn从它往上含源换线程上游视角。5. 完整可运行样例样例 1内置调度器线程名一览importreactor.core.publisher.Flux;importreactor.core.scheduler.Schedulers;publicclassBuiltinSchedulers{publicstaticvoidmain(String[]args)throwsInterruptedException{show(immediate,Schedulers.immediate());show(single,Schedulers.single());show(boundedElastic,Schedulers.boundedElastic());show(parallel,Schedulers.parallel());Thread.sleep(300);}staticvoidshow(Stringname,reactor.core.scheduler.Schedulers){Flux.range(1,1).publishOn(s).doOnNext(i-System.out.println(name - 线程: Thread.currentThread().getName())).blockLast();}}预期输出顺序可能因调度略有差异线程名前缀稳定immediate - 线程: main single - 线程: single-1 boundedElastic - 线程: boundedElastic-1 parallel - 线程: parallel-1注意immediate在当前main线程——它不开新线程只是「就在调用线程跑」。样例 2位置语义最直观对照importreactor.core.publisher.Flux;importreactor.core.scheduler.Schedulers;publicclassPublishOnVsSubscribeOn{staticintlog(Stringtag,inti){System.out.println(tag Thread.currentThread().getName() 值i);returni*10;}publicstaticvoidmain(String[]args)throwsInterruptedException{System.out.println( publishOn逐个切换下游线程 );Flux.range(1,3).map(i-log(map1(源端),i))// 订阅线程main.publishOn(Schedulers.parallel()).map(i-log(map2(parallel后),i))// parallel 线程.publishOn(Schedulers.boundedElastic()).map(i-log(map3(elastic后),i))// boundedElastic 线程.blockLast();System.out.println(\n subscribeOn决定订阅/源发生的线程 );Flux.range(1,3).subscribeOn(Schedulers.parallel())// 订阅源都搬到 parallel.map(i-log(A(parallel订阅),i))// parallel 线程.map(i-log(B,i))// 仍在 parallel.blockLast();System.out.println(\n 多个 subscribeOn最靠近 source 的生效 );Flux.range(1,3).subscribeOn(Schedulers.parallel())// 最靠近 source.map(i-log(X,i))// 在 parallel.subscribeOn(Schedulers.boundedElastic())// 下游被上游覆盖.blockLast();Thread.sleep(200);}}预期输出 publishOn逐个切换下游线程 map1(源端) main 值1 map1(源端) main 值2 map1(源端) main 值3 map2(parallel后) parallel-1 值10 map2(parallel后) parallel-1 值20 map2(parallel后) parallel-1 值30 map3(elastic后) boundedElastic-1 值100 map3(elastic后) boundedElastic-1 值200 map3(elastic后) boundedElastic-1 值300 subscribeOn决定订阅/源发生的线程 A(parallel订阅) parallel-1 值1 A(parallel订阅) parallel-1 值2 A(parallel订阅) parallel-1 值3 B parallel-1 值10 B parallel-1 值20 B parallel-1 值30 多个 subscribeOn最靠近 source 的生效 X parallel-1 值1 X parallel-1 值2 X parallel-1 值3关键观察最后的X跑在parallel而非boundedElastic——证明「多个 subscribeOn最靠近 source 的赢」。样例 3自定义命名线程池 newBoundedElasticimportreactor.core.publisher.Flux;importreactor.core.scheduler.Schedulers;importreactor.core.scheduler.Scheduler;importjava.util.concurrent.ExecutorService;importjava.util.concurrent.Executors;importjava.util.concurrent.ThreadFactory;importjava.util.concurrent.atomic.AtomicInteger;publicclassCustomScheduler{publicstaticvoidmain(String[]args)throwsInterruptedException{// 1) 自定义命名固定线程池接入既有池 / 排查友好ThreadFactorytfnewThreadFactory(){privatefinalAtomicIntegernnewAtomicInteger(1);publicThreadnewThread(Runnabler){ThreadtnewThread(r,biz-pool-n.getAndIncrement());t.setDaemon(true);returnt;}};ExecutorServicepoolExecutors.newFixedThreadPool(4,tf);SchedulercustomPoolSchedulers.fromExecutorService(pool);System.out.println( 自定义固定线程池fromExecutorService);Flux.range(1,6).publishOn(customPool).doOnNext(i-System.out.println(线程: Thread.currentThread().getName() 值i)).blockLast();pool.shutdown();// 用普通线程池记得关掉// 2) Reactor 自带 newBoundedElastic带上限、TTL、命名用完 disposeSchedulerelasticSchedulers.newBoundedElastic(8,// 最多 8 个线程1000,// 任务队列上限 1000my-elastic,60,// 空闲 60s 回收false);System.out.println(\n newBoundedElastic );Flux.range(1,4).publishOn(elastic).doOnNext(i-System.out.println(线程: Thread.currentThread().getName() 值i)).blockLast();elastic.dispose();// 释放避免线程泄漏Thread.sleep(200);}}预期输出 自定义固定线程池fromExecutorService 线程: biz-pool-1 值1 线程: biz-pool-2 值2 线程: biz-pool-3 值3 线程: biz-pool-4 值4 线程: biz-pool-1 值5 线程: biz-pool-2 值6 newBoundedElastic 线程: my-elastic-1 值1 线程: my-elastic-2 值2 线程: my-elastic-3 值3 线程: my-elastic-4 值4线程名前缀biz-pool-/my-elastic-是由你的ThreadFactory/name参数决定的——线上排查一眼可定位业务线。样例 4实战组合IO 用弹性池、计算用并行池importreactor.core.publisher.Mono;importreactor.core.scheduler.Schedulers;publicclassIoThenCompute{staticStringblockingIo(intid){try{Thread.sleep(200);}catch(InterruptedExceptionignored){}returnio-result-id;}staticintcompute(Strings){returns.hashCode();}publicstaticvoidmain(String[]args){StringresultMono.fromCallable(()-blockingIo(1)).subscribeOn(Schedulers.boundedElastic())// ① 阻塞 IO 放进弹性池.map(IoThenCompute::compute)// ② 仍在弹性池IO 刚结束.publishOn(Schedulers.parallel())// ③ 切到并行池做 CPU 计算.map(h-hashh).block();System.out.println(done - result);}}预期输出done - hash逻辑值这条链是 WebFlux / R2DBC 业务里最经典的写法① 把阻塞调用用subscribeOn(boundedElastic)搬离 EventLoop③ 计算阶段publishOn(parallel)利用多核。6. 常见坑在 EventLoop /main上阻塞响应式框架WebFlux的 IO 线程是 EventLoop上面sleep/DB 调用会拖垮所有请求。任何阻塞都要subscribeOn(boundedElastic)。parallel上做阻塞 IOparallel线程数CPU 核数被阻塞就再无线程做计算了 → 全服务卡死。阻塞一律用boundedElastic。boundedElastic上跑重计算弹性池线程多跑纯计算既浪费又可能和其他 IO 争抢。计算用parallel。每次请求都newBoundedElastic/newParallel会不断新建线程池、泄漏线程。应声明为static final常量复用或明确dispose()释放。publishOn/subscribeOn位置写反想切「处理线程」却写了subscribeOn它只管源想切「源线程」却写了publishOn。记住 4.1 / 4.2 的位置语义。以为subscribeOn能多线程并行发元素subscribeOn只是把订阅搬到单条线程不会把一个 Flux 并行化处理那要用parallel()操作符或flatMap的concurrency。7. 速查表收藏级你想做用哪个纯计算、并行处理Schedulers.parallel()publishOn阻塞 IODB/HTTP/文件Schedulers.boundedElastic()subscribeOn严格串行某类任务Schedulers.single()/newSingle(name)接入已有线程池Schedulers.fromExecutorService(pool)给业务线隔离独立池Schedulers.newBoundedElastic(...)/newParallel(...)切换「下游处理」线程publishOn(scheduler)切换「源/订阅」线程subscribeOn(scheduler)给线程命名便于排查自定义ThreadFactory或name参数8. 所以呢调度不是「锦上添花」是「正确性」在响应式框架里阻塞放错线程 生产事故。parallel算、boundedElastic堵是铁律。自定义调度器用于「隔离 可排查」给线程池命名、给不同业务线独立池线上出问题看线程 dump 能秒定位。语法就是Schedulers.fromExecutorService(pool)或Schedulers.newBoundedElastic(...)。publishOn管下游、subscribeOn管源位置一错看到的现象完全两样——务必跑一遍第 5 节样例把「线程名」印在脑子里。
