很多做 Java 和 Android 开发的朋友在刚接触响应式编程的时候都喜欢把 Mono 和 Flux 当成“更优雅的 RxJava”来用。用得多了难免会好奇一件事当我调用mono.subscribe()的时候底层到底发生了什么为什么我在 subscribe 之前写的逻辑不会执行而在 subscribe 之后写的逻辑却能跑起来这个“订阅即触发”的魔法究竟是谁在哪一行代码里实现的恰好我这段时间在做一个基于响应式流的中间件项目把 Reactive Streams 的核心接口和 Reactor 的几个关键实现类翻来覆去读了几遍。借着这篇源码分析我想把Mono#subscribe这条调用链完整地拆开来看一遍。阅读这篇文章你能搞清楚subscribe()方法的重载路由、subscribeOrReturn这条捷径存在的意义、onSubscribe阶段究竟做了什么、SET_ONCE状态标记如何防止重复订阅、以及serialize()方法为什么能保证线程安全。这篇文章适合已经会用 Mono/Flux、但想深入理解其内部机制的开发者阅读也适合那些在排查“订阅不生效”或“回调重复执行”问题时需要从源码层面找答案的工程师。1. 核心链路拆解从 subscribe() 重载到 onSubscribe 分发Mono 作为 Reactor 中表示“最多发出 0 或 1 个元素”的流其订阅入口并不像表面看起来那么简单。我第一次翻开Mono#subscribe的源码时最大的感受是这一层方法拆得很细目的就是引导调用者走向不同的处理路径避免在单一方法里堆积太多判断逻辑。1.1 门面方法两个最常用的 subscribe 重载我们先从最常用的两个方法入口看起。在Mono类中subscribe()的重载非常多但大多数最终都会汇聚到两个核心方法上public final Disposable subscribe() { return subscribeWith(new LambdaMonoSubscriber(null, null, null, null)); } public final Disposable subscribe(Consumer? super T consumer) { return subscribeWith(new LambdaMonoSubscriber(consumer, null, null, null)); }subscribe()无参版本创建的LambdaMonoSubscriber是 Reactor 为 lambda 表达式风格订阅专门设计的订阅者包装类。四个参数分别是onNext、onError、onComplete和onSubscription回调。无参调用时四个回调全部为空只有onSubscribe内部会触发 request 请求。这里有一个值得注意的点subscribeWith(Subscriber)方法才是所有订阅路径的真正汇聚点。我们继续往下追public final E extends Subscriber? super T E subscribeWith(E subscriber) { subscribe(subscriber); return subscriber; }subscribeWith做的事情非常纯粹——调用subscribe(Subscriber)然后把同一个 subscriber 原样返回。这样调用者就能在订阅的同时拿到 subscriber 引用方便后续手动调用request(n)或者cancel()。如果你写过source.subscribeWith(new BaseSubscriber())那么这里就是它返回 subscriber 实例的实现位置。1.2 双轨设计subscribeOrReturn 与 onSubscribe 的分工接下来是整个订阅流程中我认为最精巧的设计——subscribe(Subscriber)方法内部的双轨判断public final void subscribe(Subscriber? super T actual) { CoreSubscriber? super T subscriber Operators.toCoreSubscriber(actual); try { currentContext subscriber.currentContext(); } catch (Throwable throwable) { Operators.onErrorDropped(throwable, currentContext); return; } try { Subscriber? super T subscriber2 subscribeOrReturn(subscriber); if (subscriber2 ! null) { subscriber2.onSubscribe(new MonoSubscription(subscriber2)); return; } } catch (Throwable throwable) { reportSubscriptionError(throwable, subscriber); return; } onSubscribe(subscriber); }第一阶段调用subscribeOrReturn(subscriber)。这个方法允许具体的 Mono 实现类在订阅阶段直接返回一个“替代订阅者”。如果返回非 null框架就使用MonoSubscription包装这个替代订阅者并直接调用它的onSubscribe方法完成订阅整个流程到此结束。如果subscribeOrReturn返回 null说明这个 Mono 实现不打算走捷径于是进入第二阶段——调用本类的onSubscribe(subscriber)方法走标准的订阅流程。这个双轨设计解决了一个很实际的问题有很多特殊的 Mono 实现比如空的 Mono、只包含常量的 Mono根本不需要复杂的订阅机制。它们可以在subscribeOrReturn阶段直接把一个已经终结的订阅者返回省去了后续的订阅状态管理。以空 Mono 为例它的subscribeOrReturn内部会直接返回一个CancelledSubscriber这样调用方拿到的就是一个已经处于终止状态的订阅关系。排查小技巧如果你在看源码的时候找不到某个自定义 Mono 的订阅逻辑优先翻它的subscribeOrReturn方法。很多“看似没走正常订阅流程”的行为其实都在这个方法里提前处理了。1.3 LambdaMonoSubscriber四个回调一次打包LambdaMonoSubscriber是整个订阅链路的末端消费者。它把用户在subscribe()调用时传入的 lambda 表达式打包成内部字段然后在流式事件到达时逐一触发。这个类的封装修得很细致onNextConsumer处理元素到达事件。onErrorConsumer处理错误事件。onCompleteConsumer处理完成事件。onSubscriptionConsumer处理订阅建立事件可以在订阅建立时拿到Subscription对象自主控制请求策略。LambdaMonoSubscriber的onSubscribe方法会先调用用户传入的onSubscriptionConsumer如果存在然后执行s.request(Long.MAX_VALUE)。这个request调用非常关键——它告诉上游“我准备好接收数据了”从而触发整个响应式链路的运行。2. 两项核心机制PhaseChecker 与 SET_ONCE 状态标记如果你已经跟到onSubscribe(subscriber)这一步就会发现这里的实现一点也不简单。它包含了两项在我看来是 Reactor 线程安全和状态管理基石的东西PhaseChecker和SET_ONCE。2.1 从 Subscriber 到 currentThread谁在背后做事在onSubscribe方法中第一步不是直接调用订阅者的onSubscribe方法而是先从订阅者身上提取一个叫currentThread的字段void onSubscribe(Subscriber? super T actual) { currentThread (Thread)VolatileHelper.get(actual, CURRENT_THREAD); try { PhaseChecker.check(actual, SUBSCRIBE, POST_HANDLE); } catch (Throwable throwable) { currentThread null; reportSubscriptionError(throwable, actual); return; } actual.onSubscribe(this); }currentThread字段的定义是通过VarHandle或AtomicReferenceFieldUpdater之类的机制实现的本质上是订阅者上的一个 volatile 字段引用。它记录了“当前正在哪个线程上执行订阅流程”。这个字段非常关键因为它直接决定了后续数据流在哪个线程上运行。这里用VolatileHelper.get而不是直接访问字段是为了在不改变CoreSubscriber接口结构的前提下以最少的侵入方式读取订阅者内部的线程信息。这种“通过外部工具类访问内部字段”的写法在 Reactor 内部非常常见好处是字段的访问逻辑可以集中管理便于优化和调试。2.2 订阅阶段状态机四个阶段的流转逻辑PhaseChecker是订阅状态验证的守门员。它的核心职责是在onSubscribe调用前后检查当前订阅者所处的阶段是否合法。整个订阅周期被划分成四个阶段阶段名称说明典型应用POST_HANDLE订阅已建立可以执行后续操作正常数据流传输RETRY重试阶段失败重试场景PENDING等待订阅完成阶段防止循环调用NEW新订阅初始阶段所有订阅的起点check方法的逻辑非常严谨。在NEW阶段它会先检查currentThread是否已经有值。如果已经被赋值说明订阅流程已经在其他线程上开始了当前线程不能重复执行如果没有值说明当前线程是第一个到达的线程可以继续执行。在POST_HANDLE阶段逻辑正好相反——它要求currentThread必须为 null确保不会出现两个线程同时处理同一个订阅的情况。这种“两次检查、双重确认”的机制保证了订阅操作只能由单个线程完成。2.3 SET_ONCE 的巧妙之处一个整数承载两种状态SET_ONCE是 subscribe 流程中最容易出现理解偏差的地方。我第一次看到这个单词时以为它只是一个简单的“是否已经设置”的布尔标记。实际上它是一个通过整数位运算实现的多状态标记static final int ON_SUBSCRIBE_LOAD 1; static final int ON_SUBSCRIBE_CALLED 2; static final int FULL_MASK ON_SUBSCRIBE_LOAD | ON_SUBSCRIBE_CALLED;ON_SUBSCRIBE_LOAD表示onSubscribe的调用已经被加载也就是准备调用。ON_SUBSCRIBE_CALLED表示onSubscribe方法已经被实际调用。FULL_MASK两个标记位同时置 1表示onSubscribe既被加载又被调用了。为什么要拆成两个独立的位因为在onSubscribe的实现过程中先设置ON_SUBSCRIBE_LOAD再调用订阅者的onSubscribe方法最后设置ON_SUBSCRIBE_CALLED。这样即使onSubscribe执行过程中抛出了异常也能通过标记位判断出异常发生的时机。serialize()方法内部会使用这个状态标记来防止重复订阅。如果FULL_MASK已经设置说明订阅已经完成再次调用subscribe就会抛出异常或者被静默忽略避免了数据流被重复消费。3. 两处精巧设计StrictSubscriber 与 tryOnSubscribe 的边界处理如果说前两节讲的都是 subscribe 的“主干道”那这一节要聊的两处设计就是“立交桥”——它们负责在不同线程、不同订阅关系之间安全地切换和转发。这也是我在实际项目中踩坑最多、最后读源码才真正搞明白的地方。3.1 serialize() 到底在保护什么serialize()方法在 Reactive Streams 规范中承担着非常重要的职责。规范 1.3 明确规定onNext、onError、onComplete这三个方法不能并行调用也不能在onSubscribe之前调用。但在多线程环境下上游可能从多个线程同时发出信号所以serialize()要做的就是把并发信号串行化。Mono#serialize()的实现方式并不是直接修改当前 Mono而是创建了一个StrictSubscriber来包装真实的订阅者public final MonoT serialize() { return new MonoSource(this, (source, actual) - { StrictSubscriberT strictSubscriber new StrictSubscriber(actual); source.subscribe(strictSubscriber); return strictSubscriber; }); }这段代码的含义是当新的订阅者订阅serialize()返回的 Mono 时先创建一个StrictSubscriber让它订阅上游然后把真实的订阅者交给StrictSubscriber管理。这样所有来自上游的信号都先到达StrictSubscriber再由它来决定如何转发给真实的订阅者。StrictSubscriber内部用LockHelper实现了一个精简的锁机制。它的onSubscribe方法会记录当前线程后续的onNext、onError、onComplete都会先尝试获取锁获取成功后才执行回调。如果锁已经被其他线程持有当前信号就会被排队等待从而保证了信号的有序性。实战经验如果你的自定义订阅者中出现了“偶发性的回调顺序错乱”不要急着在业务代码里加锁。先检查上游是否需要serialize()。很多时候问题根本不在你的订阅者而在于上游多线程发射信号时缺少串行化保护。3.2 StrictSubscriber 转发时的两种状态StrictSubscriber在转发信号时会区分两种状态普通执行状态和可重入执行状态。这个设计非常讨巧它允许在“首次订阅”这一特定的、保证单线程的场景下直接执行回调而不用加锁只有在真正存在并发风险的重入场景下才走锁逻辑。Override public void onNext(T t) { if (StrictSubscriber.this.initialState) { StrictSubscriber.this.actual.onNext(t); } else { synchronized (this) { StrictSubscriber.this.actual.onNext(t); } } }判断依据是initialState这个布尔字段。它在订阅建立时被设为 true在第一次信号转发完成后被设为 false。这个优化能在绝大多数单线程场景下避免不必要的锁开销只有在真正需要时才启用同步机制。这种“先快路径、后慢路径”的思路在 Reactor 内部随处可见。理解了这一点你再去看其他响应式库的源码就会有一种“天下武功殊途同归”的感觉。3.3 tryOnSubscribe 的返回值一个决定去向的布尔值tryOnSubscribe是CoreSubscriber中一个容易被忽略的方法。它在serialize()和subscribe()流程中承担着边界判定的职责。与普通onSubscribe方法不同tryOnSubscribe有一个返回值default boolean tryOnSubscribe(Subscription subscription) { onSubscribe(subscription); return true; }返回true表示订阅已经成功建立。返回false则表示当前的订阅关系尚未准备好比如订阅者已经处于终止状态调用方需要采取其他措施。在LambdaMonoSubscriber中tryOnSubscribe的实现会先检查自身状态。如果订阅者已经收到过终止信号就直接返回false由上游来处理后续逻辑如果状态正常就正常调用onSubscribe并返回true。这个返回值的使用场景非常精准它让上游能够区分“订阅成功”和“订阅被拒绝”两种情况从而避免在订阅者已经终止的情况下继续下发数据。我在排查一个“订阅者收到重复数据”的 bug 时就是靠跟踪tryOnSubscribe的返回值才找到问题根源的——上游在下发数据前没有检查订阅者状态数据的request信号被重复消费了。4. 异常路径与调度入口fail() 和 run() 的设计逻辑正常路径看完之后我们再看两条“特殊路径”。这两条路径分别处理了“订阅失败”和“调度执行”两个场景它们在源码阅读中很容易被跳过但偏偏是排查问题时的关键点。4.1 fail() 方法错误报告的一次性保证在订阅流程中如果subscribeOrReturn阶段抛出了异常框架会调用reportSubscriptionError来上报错误。如果onSubscribe阶段异常则会进入fail()方法。fail()方法的设计目标很明确确保同一个错误只被报告一次。final void fail(Throwable throwable) { if (invalid) { return; } invalid true; currentContext Context.empty(); onError(throwable); }这里的关键是invalid这个字段。它通过原子操作AtomicBoolean或VarHandle保证了并发场景下的安全。当多个线程同时调用fail()时只有第一个线程能成功将invalid设为true后续线程的调用会直接返回。这样一来无论上游多么混乱订阅者都只会收到一次onError信号。这看起来很简单但它解决的是一个非常现实的并发问题。如果没有这个状态位在多线程环境下同一个异常可能会被上报多次下游的onError回调就会被重复触发导致资源清理代码重复执行。React 的commit阶段也使用类似的“一次提交”机制思路完全一致。4.2 run() 方法调度器如何触发订阅回调如果你仔细翻过Mono的源码可能注意到一个细节很多subscribeOn或publishOn相关的实现最终的入口都会落到run()方法上。这个方法是Mono内部用于“延迟订阅”的关键方法。以subscribeOn为例它的实现思路是先创建一个MonoSubscribeOn对象当订阅发生时把真实的订阅动作封装成一个Runnable交给调度器执行。这个Runnable的run()方法内部调用的是source.subscribe(actual)——也就是真正的订阅逻辑。那么在延迟订阅的场景下run()方法是怎么拿到回调结果的呢答案是OutputSubscriber。它作为一个中间层感知到上游的终止信号后会把结果转发给下游。这个设计规避了一个潜在问题如果直接把订阅逻辑放到调度线程中执行回调结果也需要在线程之间传递增加了复杂度。而通过OutputSubscriber包装回调的触发和传递都在OutputSubscriber内部完成调度线程只需要负责触发订阅动作。个人心得排查线上问题时如果发现回调没有在预期线程触发优先看调度器是否进入了run()方法。我就遇过一次“回调一直在非预期线程”的问题最后发现是run()方法里的调度器按了两次调度导致回调被分发到了错误线程。4.3 从 subscribe 到实际执行一整个流程的串联如果把上面所有环节串联起来一个完整的mono.subscribe()调用流程大概是这样的用户调用subscribe(Consumer)创建LambdaMonoSubscriber。进入subscribe(Subscriber)尝试从订阅者获取currentContext和currentThread。调用subscribeOrReturn如果返回非 null直接通过MonoSubscription完成订阅。如果返回 null调用onSubscribe(subscriber)创建订阅连接。在onSubscribe内部通过PhaseChecker验证订阅状态调用订阅者的onSubscribe(this)。订阅者LambdaMonoSubscriber在onSubscribe中调用request(Long.MAX_VALUE)触发上游数据发射。数据流经onNext到达订阅者最终触发用户的Consumer回调。这 7 步中每一步都有若干边界情况的处理上游为空、订阅者已取消、并发线程竞争、异常抛出等等。一个简单的subscribe()调用背后是一整套完整的逻辑保障。5. 常见问题与排查技巧实录在读源码和实际项目的过程中我积累了一些和subscribe相关的排查经验这里整理成几个典型问题供大家参考。5.1 订阅后回调未执行可能卡在哪一步最典型的场景是代码里调用了subscribe()但回调始终没有执行。按我的排查顺序通常按以下步骤定位检查是否有subscribeOn或publishOn调度器介入。调度器如果配置错误订阅动作可能在调度线程中被排队导致延迟执行或永不执行。检查subscribeOrReturn是否返回了“非 null”的终结订阅者。如果上游是空 Mono订阅可能直接走捷径完成而不会触发业务回调。检查订阅者是否在onSubscribe中没有调用request(n)。如果上游需要主动请求才能发射数据而订阅者没有请求数据流就会挂起。最后检查是否发生了异常并被静默上报onErrorDropped。这可以通过设置Hooks.onErrorDropped回调来捕获。5.2 回调被重复执行大概率是状态位没设置回调被重复执行的情况通常发生在多线程并发订阅同一 Mono 时。核心原因是subscribe流程中依赖的状态位如SET_ONCE、invalid在并发场景下由于竞态条件而没有生效。排查时重点关注订阅者是否实现了CoreSubscriber以及onSubscribe方法中是否调用了tryOnSubscribe。如果tryOnSubscribe返回false而没有正确处理就可能出现重复调用。5.3 线程切换不符合预期看看 currentThread 做了哪些事currentThread字段在onSubscribe阶段被赋值后续的数据流发射和信号传递都会基于这个字段进行线程归属判断。如果你发现回调执行的线程不符合预期可以检查currentThread是在哪个线程被赋值的。一个很常见的坑是在subscribeOn调度的场景下currentThread会被赋值为调度线程而不是调用subscribe()的业务线程。这在语义上是正确的——因为真正的订阅发生在调度线程上。如果业务代码对线程有强依赖就要在回调内部显式检查Thread.currentThread()。6. 写在最后的阅读建议我现在回头复盘这段阅读经历最有价值的收获并不是记住了某个具体方法的行为而是建立起了一套理解响应式框架的思维框架一个看似简单的操作背后一定有一系列状态位、阶段校验和边界处理来保障其正确性。如果你是第一次接触 Reactor 的源码我建议你先从LambdaMonoSubscriber入手顺着一条最简单的订阅链走一遍把每个方法的调用顺序和参数传递关系画成图。有了整体印象之后再去细抠SET_ONCE、PhaseChecker这些并发控制机制会轻松很多。后续如果你想继续深入还可以沿着这条线去看Flux的订阅流程、Operators工具类中的各类lift操作以及Schedulers调度器的实现。你会发现虽然类名不同但设计思路是高度一致的——先设置状态再调用回调最后处理异常。当你能在阅读任何一段响应式代码时都下意识地寻找这三个步骤你就算真正入门了。
很多做 Java 和 Android 开发的朋友在刚接触响应式编程的时候都喜欢把 Mono 和 Flux 当成“更优雅的 RxJava”来用。用得多了难免会好奇一件事当我调用mono.subscribe()的时候底层到底发生了什么为什么我在 subscribe 之前写的逻辑不会执行而在 subscribe 之后写的逻辑却能跑起来这个“订阅即触发”的魔法究竟是谁在哪一行代码里实现的恰好我这段时间在做一个基于响应式流的中间件项目把 Reactive Streams 的核心接口和 Reactor 的几个关键实现类翻来覆去读了几遍。借着这篇源码分析我想把Mono#subscribe这条调用链完整地拆开来看一遍。阅读这篇文章你能搞清楚subscribe()方法的重载路由、subscribeOrReturn这条捷径存在的意义、onSubscribe阶段究竟做了什么、SET_ONCE状态标记如何防止重复订阅、以及serialize()方法为什么能保证线程安全。这篇文章适合已经会用 Mono/Flux、但想深入理解其内部机制的开发者阅读也适合那些在排查“订阅不生效”或“回调重复执行”问题时需要从源码层面找答案的工程师。1. 核心链路拆解从 subscribe() 重载到 onSubscribe 分发Mono 作为 Reactor 中表示“最多发出 0 或 1 个元素”的流其订阅入口并不像表面看起来那么简单。我第一次翻开Mono#subscribe的源码时最大的感受是这一层方法拆得很细目的就是引导调用者走向不同的处理路径避免在单一方法里堆积太多判断逻辑。1.1 门面方法两个最常用的 subscribe 重载我们先从最常用的两个方法入口看起。在Mono类中subscribe()的重载非常多但大多数最终都会汇聚到两个核心方法上public final Disposable subscribe() { return subscribeWith(new LambdaMonoSubscriber(null, null, null, null)); } public final Disposable subscribe(Consumer? super T consumer) { return subscribeWith(new LambdaMonoSubscriber(consumer, null, null, null)); }subscribe()无参版本创建的LambdaMonoSubscriber是 Reactor 为 lambda 表达式风格订阅专门设计的订阅者包装类。四个参数分别是onNext、onError、onComplete和onSubscription回调。无参调用时四个回调全部为空只有onSubscribe内部会触发 request 请求。这里有一个值得注意的点subscribeWith(Subscriber)方法才是所有订阅路径的真正汇聚点。我们继续往下追public final E extends Subscriber? super T E subscribeWith(E subscriber) { subscribe(subscriber); return subscriber; }subscribeWith做的事情非常纯粹——调用subscribe(Subscriber)然后把同一个 subscriber 原样返回。这样调用者就能在订阅的同时拿到 subscriber 引用方便后续手动调用request(n)或者cancel()。如果你写过source.subscribeWith(new BaseSubscriber())那么这里就是它返回 subscriber 实例的实现位置。1.2 双轨设计subscribeOrReturn 与 onSubscribe 的分工接下来是整个订阅流程中我认为最精巧的设计——subscribe(Subscriber)方法内部的双轨判断public final void subscribe(Subscriber? super T actual) { CoreSubscriber? super T subscriber Operators.toCoreSubscriber(actual); try { currentContext subscriber.currentContext(); } catch (Throwable throwable) { Operators.onErrorDropped(throwable, currentContext); return; } try { Subscriber? super T subscriber2 subscribeOrReturn(subscriber); if (subscriber2 ! null) { subscriber2.onSubscribe(new MonoSubscription(subscriber2)); return; } } catch (Throwable throwable) { reportSubscriptionError(throwable, subscriber); return; } onSubscribe(subscriber); }第一阶段调用subscribeOrReturn(subscriber)。这个方法允许具体的 Mono 实现类在订阅阶段直接返回一个“替代订阅者”。如果返回非 null框架就使用MonoSubscription包装这个替代订阅者并直接调用它的onSubscribe方法完成订阅整个流程到此结束。如果subscribeOrReturn返回 null说明这个 Mono 实现不打算走捷径于是进入第二阶段——调用本类的onSubscribe(subscriber)方法走标准的订阅流程。这个双轨设计解决了一个很实际的问题有很多特殊的 Mono 实现比如空的 Mono、只包含常量的 Mono根本不需要复杂的订阅机制。它们可以在subscribeOrReturn阶段直接把一个已经终结的订阅者返回省去了后续的订阅状态管理。以空 Mono 为例它的subscribeOrReturn内部会直接返回一个CancelledSubscriber这样调用方拿到的就是一个已经处于终止状态的订阅关系。排查小技巧如果你在看源码的时候找不到某个自定义 Mono 的订阅逻辑优先翻它的subscribeOrReturn方法。很多“看似没走正常订阅流程”的行为其实都在这个方法里提前处理了。1.3 LambdaMonoSubscriber四个回调一次打包LambdaMonoSubscriber是整个订阅链路的末端消费者。它把用户在subscribe()调用时传入的 lambda 表达式打包成内部字段然后在流式事件到达时逐一触发。这个类的封装修得很细致onNextConsumer处理元素到达事件。onErrorConsumer处理错误事件。onCompleteConsumer处理完成事件。onSubscriptionConsumer处理订阅建立事件可以在订阅建立时拿到Subscription对象自主控制请求策略。LambdaMonoSubscriber的onSubscribe方法会先调用用户传入的onSubscriptionConsumer如果存在然后执行s.request(Long.MAX_VALUE)。这个request调用非常关键——它告诉上游“我准备好接收数据了”从而触发整个响应式链路的运行。2. 两项核心机制PhaseChecker 与 SET_ONCE 状态标记如果你已经跟到onSubscribe(subscriber)这一步就会发现这里的实现一点也不简单。它包含了两项在我看来是 Reactor 线程安全和状态管理基石的东西PhaseChecker和SET_ONCE。2.1 从 Subscriber 到 currentThread谁在背后做事在onSubscribe方法中第一步不是直接调用订阅者的onSubscribe方法而是先从订阅者身上提取一个叫currentThread的字段void onSubscribe(Subscriber? super T actual) { currentThread (Thread)VolatileHelper.get(actual, CURRENT_THREAD); try { PhaseChecker.check(actual, SUBSCRIBE, POST_HANDLE); } catch (Throwable throwable) { currentThread null; reportSubscriptionError(throwable, actual); return; } actual.onSubscribe(this); }currentThread字段的定义是通过VarHandle或AtomicReferenceFieldUpdater之类的机制实现的本质上是订阅者上的一个 volatile 字段引用。它记录了“当前正在哪个线程上执行订阅流程”。这个字段非常关键因为它直接决定了后续数据流在哪个线程上运行。这里用VolatileHelper.get而不是直接访问字段是为了在不改变CoreSubscriber接口结构的前提下以最少的侵入方式读取订阅者内部的线程信息。这种“通过外部工具类访问内部字段”的写法在 Reactor 内部非常常见好处是字段的访问逻辑可以集中管理便于优化和调试。2.2 订阅阶段状态机四个阶段的流转逻辑PhaseChecker是订阅状态验证的守门员。它的核心职责是在onSubscribe调用前后检查当前订阅者所处的阶段是否合法。整个订阅周期被划分成四个阶段阶段名称说明典型应用POST_HANDLE订阅已建立可以执行后续操作正常数据流传输RETRY重试阶段失败重试场景PENDING等待订阅完成阶段防止循环调用NEW新订阅初始阶段所有订阅的起点check方法的逻辑非常严谨。在NEW阶段它会先检查currentThread是否已经有值。如果已经被赋值说明订阅流程已经在其他线程上开始了当前线程不能重复执行如果没有值说明当前线程是第一个到达的线程可以继续执行。在POST_HANDLE阶段逻辑正好相反——它要求currentThread必须为 null确保不会出现两个线程同时处理同一个订阅的情况。这种“两次检查、双重确认”的机制保证了订阅操作只能由单个线程完成。2.3 SET_ONCE 的巧妙之处一个整数承载两种状态SET_ONCE是 subscribe 流程中最容易出现理解偏差的地方。我第一次看到这个单词时以为它只是一个简单的“是否已经设置”的布尔标记。实际上它是一个通过整数位运算实现的多状态标记static final int ON_SUBSCRIBE_LOAD 1; static final int ON_SUBSCRIBE_CALLED 2; static final int FULL_MASK ON_SUBSCRIBE_LOAD | ON_SUBSCRIBE_CALLED;ON_SUBSCRIBE_LOAD表示onSubscribe的调用已经被加载也就是准备调用。ON_SUBSCRIBE_CALLED表示onSubscribe方法已经被实际调用。FULL_MASK两个标记位同时置 1表示onSubscribe既被加载又被调用了。为什么要拆成两个独立的位因为在onSubscribe的实现过程中先设置ON_SUBSCRIBE_LOAD再调用订阅者的onSubscribe方法最后设置ON_SUBSCRIBE_CALLED。这样即使onSubscribe执行过程中抛出了异常也能通过标记位判断出异常发生的时机。serialize()方法内部会使用这个状态标记来防止重复订阅。如果FULL_MASK已经设置说明订阅已经完成再次调用subscribe就会抛出异常或者被静默忽略避免了数据流被重复消费。3. 两处精巧设计StrictSubscriber 与 tryOnSubscribe 的边界处理如果说前两节讲的都是 subscribe 的“主干道”那这一节要聊的两处设计就是“立交桥”——它们负责在不同线程、不同订阅关系之间安全地切换和转发。这也是我在实际项目中踩坑最多、最后读源码才真正搞明白的地方。3.1 serialize() 到底在保护什么serialize()方法在 Reactive Streams 规范中承担着非常重要的职责。规范 1.3 明确规定onNext、onError、onComplete这三个方法不能并行调用也不能在onSubscribe之前调用。但在多线程环境下上游可能从多个线程同时发出信号所以serialize()要做的就是把并发信号串行化。Mono#serialize()的实现方式并不是直接修改当前 Mono而是创建了一个StrictSubscriber来包装真实的订阅者public final MonoT serialize() { return new MonoSource(this, (source, actual) - { StrictSubscriberT strictSubscriber new StrictSubscriber(actual); source.subscribe(strictSubscriber); return strictSubscriber; }); }这段代码的含义是当新的订阅者订阅serialize()返回的 Mono 时先创建一个StrictSubscriber让它订阅上游然后把真实的订阅者交给StrictSubscriber管理。这样所有来自上游的信号都先到达StrictSubscriber再由它来决定如何转发给真实的订阅者。StrictSubscriber内部用LockHelper实现了一个精简的锁机制。它的onSubscribe方法会记录当前线程后续的onNext、onError、onComplete都会先尝试获取锁获取成功后才执行回调。如果锁已经被其他线程持有当前信号就会被排队等待从而保证了信号的有序性。实战经验如果你的自定义订阅者中出现了“偶发性的回调顺序错乱”不要急着在业务代码里加锁。先检查上游是否需要serialize()。很多时候问题根本不在你的订阅者而在于上游多线程发射信号时缺少串行化保护。3.2 StrictSubscriber 转发时的两种状态StrictSubscriber在转发信号时会区分两种状态普通执行状态和可重入执行状态。这个设计非常讨巧它允许在“首次订阅”这一特定的、保证单线程的场景下直接执行回调而不用加锁只有在真正存在并发风险的重入场景下才走锁逻辑。Override public void onNext(T t) { if (StrictSubscriber.this.initialState) { StrictSubscriber.this.actual.onNext(t); } else { synchronized (this) { StrictSubscriber.this.actual.onNext(t); } } }判断依据是initialState这个布尔字段。它在订阅建立时被设为 true在第一次信号转发完成后被设为 false。这个优化能在绝大多数单线程场景下避免不必要的锁开销只有在真正需要时才启用同步机制。这种“先快路径、后慢路径”的思路在 Reactor 内部随处可见。理解了这一点你再去看其他响应式库的源码就会有一种“天下武功殊途同归”的感觉。3.3 tryOnSubscribe 的返回值一个决定去向的布尔值tryOnSubscribe是CoreSubscriber中一个容易被忽略的方法。它在serialize()和subscribe()流程中承担着边界判定的职责。与普通onSubscribe方法不同tryOnSubscribe有一个返回值default boolean tryOnSubscribe(Subscription subscription) { onSubscribe(subscription); return true; }返回true表示订阅已经成功建立。返回false则表示当前的订阅关系尚未准备好比如订阅者已经处于终止状态调用方需要采取其他措施。在LambdaMonoSubscriber中tryOnSubscribe的实现会先检查自身状态。如果订阅者已经收到过终止信号就直接返回false由上游来处理后续逻辑如果状态正常就正常调用onSubscribe并返回true。这个返回值的使用场景非常精准它让上游能够区分“订阅成功”和“订阅被拒绝”两种情况从而避免在订阅者已经终止的情况下继续下发数据。我在排查一个“订阅者收到重复数据”的 bug 时就是靠跟踪tryOnSubscribe的返回值才找到问题根源的——上游在下发数据前没有检查订阅者状态数据的request信号被重复消费了。4. 异常路径与调度入口fail() 和 run() 的设计逻辑正常路径看完之后我们再看两条“特殊路径”。这两条路径分别处理了“订阅失败”和“调度执行”两个场景它们在源码阅读中很容易被跳过但偏偏是排查问题时的关键点。4.1 fail() 方法错误报告的一次性保证在订阅流程中如果subscribeOrReturn阶段抛出了异常框架会调用reportSubscriptionError来上报错误。如果onSubscribe阶段异常则会进入fail()方法。fail()方法的设计目标很明确确保同一个错误只被报告一次。final void fail(Throwable throwable) { if (invalid) { return; } invalid true; currentContext Context.empty(); onError(throwable); }这里的关键是invalid这个字段。它通过原子操作AtomicBoolean或VarHandle保证了并发场景下的安全。当多个线程同时调用fail()时只有第一个线程能成功将invalid设为true后续线程的调用会直接返回。这样一来无论上游多么混乱订阅者都只会收到一次onError信号。这看起来很简单但它解决的是一个非常现实的并发问题。如果没有这个状态位在多线程环境下同一个异常可能会被上报多次下游的onError回调就会被重复触发导致资源清理代码重复执行。React 的commit阶段也使用类似的“一次提交”机制思路完全一致。4.2 run() 方法调度器如何触发订阅回调如果你仔细翻过Mono的源码可能注意到一个细节很多subscribeOn或publishOn相关的实现最终的入口都会落到run()方法上。这个方法是Mono内部用于“延迟订阅”的关键方法。以subscribeOn为例它的实现思路是先创建一个MonoSubscribeOn对象当订阅发生时把真实的订阅动作封装成一个Runnable交给调度器执行。这个Runnable的run()方法内部调用的是source.subscribe(actual)——也就是真正的订阅逻辑。那么在延迟订阅的场景下run()方法是怎么拿到回调结果的呢答案是OutputSubscriber。它作为一个中间层感知到上游的终止信号后会把结果转发给下游。这个设计规避了一个潜在问题如果直接把订阅逻辑放到调度线程中执行回调结果也需要在线程之间传递增加了复杂度。而通过OutputSubscriber包装回调的触发和传递都在OutputSubscriber内部完成调度线程只需要负责触发订阅动作。个人心得排查线上问题时如果发现回调没有在预期线程触发优先看调度器是否进入了run()方法。我就遇过一次“回调一直在非预期线程”的问题最后发现是run()方法里的调度器按了两次调度导致回调被分发到了错误线程。4.3 从 subscribe 到实际执行一整个流程的串联如果把上面所有环节串联起来一个完整的mono.subscribe()调用流程大概是这样的用户调用subscribe(Consumer)创建LambdaMonoSubscriber。进入subscribe(Subscriber)尝试从订阅者获取currentContext和currentThread。调用subscribeOrReturn如果返回非 null直接通过MonoSubscription完成订阅。如果返回 null调用onSubscribe(subscriber)创建订阅连接。在onSubscribe内部通过PhaseChecker验证订阅状态调用订阅者的onSubscribe(this)。订阅者LambdaMonoSubscriber在onSubscribe中调用request(Long.MAX_VALUE)触发上游数据发射。数据流经onNext到达订阅者最终触发用户的Consumer回调。这 7 步中每一步都有若干边界情况的处理上游为空、订阅者已取消、并发线程竞争、异常抛出等等。一个简单的subscribe()调用背后是一整套完整的逻辑保障。5. 常见问题与排查技巧实录在读源码和实际项目的过程中我积累了一些和subscribe相关的排查经验这里整理成几个典型问题供大家参考。5.1 订阅后回调未执行可能卡在哪一步最典型的场景是代码里调用了subscribe()但回调始终没有执行。按我的排查顺序通常按以下步骤定位检查是否有subscribeOn或publishOn调度器介入。调度器如果配置错误订阅动作可能在调度线程中被排队导致延迟执行或永不执行。检查subscribeOrReturn是否返回了“非 null”的终结订阅者。如果上游是空 Mono订阅可能直接走捷径完成而不会触发业务回调。检查订阅者是否在onSubscribe中没有调用request(n)。如果上游需要主动请求才能发射数据而订阅者没有请求数据流就会挂起。最后检查是否发生了异常并被静默上报onErrorDropped。这可以通过设置Hooks.onErrorDropped回调来捕获。5.2 回调被重复执行大概率是状态位没设置回调被重复执行的情况通常发生在多线程并发订阅同一 Mono 时。核心原因是subscribe流程中依赖的状态位如SET_ONCE、invalid在并发场景下由于竞态条件而没有生效。排查时重点关注订阅者是否实现了CoreSubscriber以及onSubscribe方法中是否调用了tryOnSubscribe。如果tryOnSubscribe返回false而没有正确处理就可能出现重复调用。5.3 线程切换不符合预期看看 currentThread 做了哪些事currentThread字段在onSubscribe阶段被赋值后续的数据流发射和信号传递都会基于这个字段进行线程归属判断。如果你发现回调执行的线程不符合预期可以检查currentThread是在哪个线程被赋值的。一个很常见的坑是在subscribeOn调度的场景下currentThread会被赋值为调度线程而不是调用subscribe()的业务线程。这在语义上是正确的——因为真正的订阅发生在调度线程上。如果业务代码对线程有强依赖就要在回调内部显式检查Thread.currentThread()。6. 写在最后的阅读建议我现在回头复盘这段阅读经历最有价值的收获并不是记住了某个具体方法的行为而是建立起了一套理解响应式框架的思维框架一个看似简单的操作背后一定有一系列状态位、阶段校验和边界处理来保障其正确性。如果你是第一次接触 Reactor 的源码我建议你先从LambdaMonoSubscriber入手顺着一条最简单的订阅链走一遍把每个方法的调用顺序和参数传递关系画成图。有了整体印象之后再去细抠SET_ONCE、PhaseChecker这些并发控制机制会轻松很多。后续如果你想继续深入还可以沿着这条线去看Flux的订阅流程、Operators工具类中的各类lift操作以及Schedulers调度器的实现。你会发现虽然类名不同但设计思路是高度一致的——先设置状态再调用回调最后处理异常。当你能在阅读任何一段响应式代码时都下意识地寻找这三个步骤你就算真正入门了。
