Reactor publishOn 线程切换的底层秘密

用户7511514

发布于 2026-07-09 15:46:54

从一个 Bug 说起："为什么加了一行 .publishOn(Schedulers.boundedElastic()) 就不输出了？"

一、现象

写了一段 Reactor 代码：

Mono<String> mono = Mono.fromCallable(() -> "Hello")
    .map(s -> s + " map")
    .publishOn(Schedulers.boundedElastic())
    .doOnNext(v -> System.out.println("值=" + v + " 线程=" + Thread.currentThread().getName()));

mono.subscribe(v -> System.out.println("最终=" + v));

去掉 publishOn 那一行，一切正常。加上它，什么都没打印。

为什么？

因为 publishOn 把下游逻辑扔到了另一个线程去执行，而 main 线程不等它结束就退出了，JVM 直接关闭。

那 publishOn 到底是怎么把代码"扔到另一个线程"的？答案藏在这几个类的源码里。

二、MonoPublishOn — 最简洁的入口

Mono 的 publishOn 最终创建的是 MonoPublishOn，包路径：reactor.core.publisher.MonoPublishOn。

看它的核心——内部类 PublishOnSubscriber，同时实现了三个角色：

staticfinalclassPublishOnSubscriber<T>
implementsInnerOperator<T, T>, Runnable{
// InnerOperator = Subscriber（收数据）
// Runnable = 可被线程池执行

这个类既负责收上游的数据，又负责把自己提交给线程池，还在 run() 里转发数据给下游。一个对象撑起了整个线程切换。

三、四步理解线程切换

第一步：onNext — 存数据，提交任务

// MonoPublishOn.java 第 118 行
public void onNext(T t){
    value = t;                            // ① 存下唯一的值
    trySchedule(this, null, t);           // ② 提交"自己"到线程池
}

这里只做了两件极轻的事：

把上游发来的值写进 volatile T value 字段
调用 trySchedule

几乎没有阻塞，fromCallable 所在的线程瞬间返回。这就是"异步"的来源。

第二步：trySchedule — 把自己扔进线程池

// MonoPublishOn.java 第 136 行
void trySchedule(...){
if(future != null) return;            // 已提交，防重复try {
        future = this.scheduler.schedule(this);
//                     ↑↑↑↑↑↑
//         this = PublishOnSubscriber 自己
    } catch (RejectedExecutionException ree) {
        actual.onError(...);
    }
}

关键代码就一行：scheduler.schedule(this)。

这里的 scheduler 就是你传入的 Schedulers.boundedElastic()，this 就是 PublishOnSubscriber 自己——因为它实现了 Runnable。

第三步：线程池队列 — 真正的"切换点"

scheduler.schedule(this) 经过层层委托，最终走到的代码在 Schedulers.java：

// Schedulers.java 第 1444 行
static Disposable workerSchedule(ScheduledExecutorService exec, ... , Runnable task, ...){
    WorkerTask sr = new WorkerTask(task, ...);
//                 ↑↑↑↑↑
//         task = PublishOnSubscriber

    Future<?> f;
    if (delay <= 0L) {
        f = exec.submit((Callable<?>) sr);   // ← 放进线程池的工作队列！
    }
// ...
}

exec.submit(sr) 这行代码就是线程切换的真身。

exec 是一个 ScheduledExecutorService（即线程池），它的内部有一个 BlockingQueue（工作队列）。submit 就是把 PublishOnSubscriber 这个 Runnable 对象放进这个队列里。

当线程池中有空闲线程时，会从这个队列取出 Runnable 并执行它的 run() 方法。

至此，代码已经从原线程（调用 onNext 的线程）切换到了线程池的线程。

第四步：run — 在新线程上转发数据

// MonoPublishOn.java 第 174 行
publicvoidrun(){
if (OperatorDisposables.isDisposed(future)) return;

    T v = (T) VALUE.getAndSet(this, null);  // 取出之前存的值（原子操作）

    if (v != null) {
        actual.onNext(v);                   // ← 在新线程上，调下游的 onNext
        actual.onComplete();
    } else {
        Throwable e = error;
if (e != null) actual.onError(e);
else actual.onComplete();
    }
}

run() 在线程池的线程上执行：

取数据VALUE.getAndSet(this, null) 从字段中原子取出之前存的值
转发actual.onNext(v) 调下游的 doOnNext 回调

所以你的 System.out.println("值=Hello map") 就在这里打印——线程名变成了 boundedElastic-1。

四、完整流程图

                                  线程池的工作队列
                                      ┌──────┐
                                      │task  │
                                      ├──────┤
  fromCallable 线程                    │task  │
       │                              ├──────┤
       │                              │task  │
       │                              └──────┘
       │                                  ↑
       │                                  │ poll()
       │                                  │
  ─────┴────── onNext("Hello") ──── schedule(this) ────┐
                                                     │
  main线程                                            │
                                                 线程池线程
                                               ┌─── run() ────┐
                                               │ 取出 value    │
                                               │ actual.onNext │
                                               │ 打印输出      │
                                               └──────────────┘

两层"队列"的关系：

队列	存什么	谁在用
PublishOnSubscriber.value（volatile 字段）	上游的业务数据	存：onNext / 取：run
线程池的 BlockingQueue	PublishOnSubscriber 这个 Runnable	存：exec.submit / 取：线程池核心线程

这两层是串行的：数据先存进 value，再把"取数据+转发"这个动作提交到线程池队列。

五、为什么去掉 publishOn 就正常？

没有 publishOn 时，整条链同步执行在 main 线程：

main: fromCallable → map → doOnNext → subscribe 回调 → 打印 → main 结束 ✓

加上 publishOn 后：

main: fromCallable → map → "提交任务, 返回" ─→ main 结束（JVM退出！）
                                  │
                          boundedElastic 线程: doOnNext → subscribe 回调 → 来不及打印 💀

根本原因：Schedulers.boundedElastic() 创建的线程是守护线程（daemon），JVM 不会等守护线程结束。

解决：用 block() 替代 subscribe()，或用 CountDownLatch 等待。

六、FluxPublishOn 有什么不同？

Flux 的 publishOn（FluxPublishOn）思路一样，但多了一个真正的队列：

// FluxPublishOn.java 第 214 行
public void onNext(T t){
 if (!queue.offer(t)) {                // ← 多个元素用 Queue
        error = ... QUEUE_FULL;
        done = true;
    }
    trySchedule(this, null, t);           // ← 同样提交自己
}

Flux 有多个元素，所以不能像 Mono 那样用单个字段存，而是用 Queue<T>：

	MonoPublishOn	FluxPublishOn
存数据	volatile T value（一个字段）	Queue<T> queue（队列）
取数据	VALUE.getAndSet(this, null)	queue.poll()
提交	scheduler.schedule(this)	worker.schedule(this)