啃透Mono.subscribe源码：响应式编程的触发与背压机制

1. 为什么值得啃一啃Mono的subscribe源码

先抛一个绝大多数写响应式代码的开发者都遇到过的困惑：一个Mono链路写完，没数据、没报错、甚至没打日志，整个流程静悄悄。排查半天发现，不是业务代码出了问题，而是压根没人调用subscribe()。响应式流是惰性的，定义了一整套数据处理流水线，但只有触发了subscribe，数据才开始流动。

这个触发点，就是Mono的subscribe方法。它是整个响应式链路的总开关，是所有回调的汇聚点，也是背压协议的起点。读懂它，你就真正理解了响应式编程的运行机制，而不是停留在“链式调用很优雅”的表层。

这篇文章的源码分析基于目前Java生态里比较常用的响应式库版本（Reactor 3.x系列），核心方法签名和实现思路在不同版本间基本稳定。我尽量把关键代码段贴出来，配合调用栈和执行顺序去讲，不搞逐行注释式的枯燥解读。

这个分析适合三类人：

• 写业务代码时总被响应式链路的“诡异行为”困扰，想搞清楚背后到底发生了什么。
• 准备面试，想对响应式编程原理有深度理解。
• 想通过一份优秀源码提升自己的代码设计能力。

2. 入口方法：subscribe()到底做了什么

2.1 从源码第一行开始看

Mono接口的subscribe最核心的实现，直接看Mono抽象类里的这段代码：

@Override public final Disposable subscribe( @Nullable Consumer<? super T> actual, @Nullable Consumer<? super Throwable> errorConsumer, @Nullable Runnable completeConsumer, @Nullable Consumer<? super Subscription> subscriptionConsumer) { // 省略了部分变量声明 if (actual != null) { try { // 注册钩子：每个订阅者都会被包装一层 actual = Hooks.onEachOperator(actual); } catch (Throwable t) { // 异常处理... } } // 核心调度入口 return onSubscribe(actual, errorConsumer, completeConsumer, subscriptionConsumer); }

看到onSubscribe这个方法名，很多人的第一反应是“这不就是Subscriber接口里的那个onSubscribe吗”。这里要区分清楚：Subscriber.onSubscribe(Subscription)是协议回调，由Publisher在订阅建立时回调给订阅者；而这里的Mono.onSubscribe(...)是Mono内部的一个抽象方法，接收我们传入的几个Consumer，负责真正构建订阅关系。同名不同义，别搞混了。

这段代码里还有一个容易被忽略的点：Hooks.onEachOperator。它是Reactor的全局钩子机制，允许你给整个响应式链路注入通用逻辑，比如全链路日志追踪、指标采集。每次调用subscribe，任何Consumer都会被这个钩子重新包装一次。如果你在排查性能问题时发现调用链里多了一层，大概率就是这里被某个全局Hook介入了。

2.2 四个参数的真实语义

我在实际带团队的过程中发现，很多同学对这四个参数的理解是模糊的。表面上看，它们分别对应“正常数据”“异常”“完成”“订阅关系”四个时机，但有几个细节值得注意：

特别说一下errorConsumer。源码在构建最终Subscriber时，有一个很重要的逻辑：如果errorConsumer == null，它会默认抛出Exceptions.propagate(t)——也就是把异常重新抛出来。所以如果你只传了一个Consumer而没传errorConsumer，一旦链路出错，异常不会安安静静消失，而是会在调用线程里抛出来。如果你在写subscribe(System.out::println)这样的代码时突然看到异常从奇怪的地方冒出来，就是这个原因。

2.3 subscribe返回的Disposable是怎么回事

再看方法签名，返回值是Disposable。这意味着一个Mono被订阅后，你是可以主动取消的。底层实现把所有的Consumer包成了一个LambdaMonoSubscriber，这个类同时实现了Disposable和Subscription两个接口。

Disposable接口的dispose()方法会让订阅者进入取消状态，关键代码如下：

@Override public void dispose() { if (STATE.getAndSet(this, TERMINATED) != TERMINATED) { // 如果上游还有Subscription，取消它 if (this.s != null) { this.s.cancel(); } } }

这段代码的核心是CAS（Compare-And-Swap）操作。STATE是一个AtomicReferenceFieldUpdater，它保证dispose操作的原子性和线程安全性。因为dispose()可能在任意线程被调用，如果没有CAS保护，就可能在“正在取消”和“正在处理元素”之间产生竞态条件，导致重复取消或逻辑状态错乱。

我第一次看到这个实现时最大的感悟是：原来订阅关系只是一个状态机。TERMINATED就是终态，一旦进入，任何操作都会被挡在外面。后面分析request、onNext的执行路径时，你会反复看到同一个模式——先检查状态，再执行逻辑，最后回写状态。

3. 订阅链路的建立：从subscribe到onSubscribe的完整旅程

3.1 LambdaMonoSubscriber：被包出来的订阅者

subscribe方法最终会调用Mono.subscribe(actual, errorConsumer, completeConsumer, subscriptionConsumer)，而这个方法的核心动作就是把传进来的几个Consumer组装成一个完整的Subscriber。

看Reactors的LambdaMonoSubscriber核心实现：

@Override public void onSubscribe(Subscription s) { if (this.s != null) { s.cancel(); return; } this.s = s; if (subscriptionConsumer != null) { try { subscriptionConsumer.accept(this); } catch (Throwable t) { // 异常处理 } } }

注意几个值得玩味的点。

第一，如果subscriptionConsumer为null，这里不会主动调用request。那么问题来了：没人request，上游的数据怎么发下来？

这里要理解Mono的“标准用法”。看subscribe()的另一个重载实现，内部会创建一个MonoSubscriber，这个订阅者在onSubscribe时默认会调用request(Long.MAX_VALUE)，也就是“无限请求”。所以作为使用者，通常你不传subscriptionConsumer，整个订阅链会按“有多少发多少”的方式运行。

我见过有同事在排查“为什么数据不往下走”的问题时，先在subscriptionConsumer里自定义了request(0)，然后一脸疑惑为什么什么都不触发。原因就是Mono本身按“最多一个元素”设计，你可以控制请求节奏，但request(0)就是明确的“我不接收任何数据”。

第二，注意this.s != null这个判断。如果同一个LambdaMonoSubscriber实例被两个上游Publisher同时订阅，后面的那个会被直接cancel。这在链式调用中几乎不可能发生，但如果你在复杂场景中手动复用一个Subscriber实例，这是一个重要的防重入保护。

3.2 onSubscribe之后，request从哪里“发起”

细看LambdaMonoSubscriber对Subscription接口的实现：

@Override public void request(long n) { Subscription s = this.s; if (s != null) { s.request(n); } } @Override public void cancel() { Subscription s = this.s; if (s != null) { this.s = null; s.cancel(); } }

这里的this.s是谁？是上游Publisher在调用onSubscribe时传进来的那个Subscription。它可能来自MonoJust、MonoCreate或者一个操作算子（比如map、flatMap）构造的中间Subscription。

整个订阅链条的建立，本质上是一个递归的“反向连接”过程。一个Mono.map()产生的Mono，它的subscribe会先连接到上游，等上游的Subscription准备就绪后，再逐步把数据流引回来。

这里画一个简单但关键的链路图：

调用方.subscribe(consumer) ↓ MonoMap 的 subscribe 方法 ↓ 调用 source.subscribe(新的MapSubscriber) ↓ MonoJust 的 subscribe 方法 ↓ MonoJust 直接调用 actual.onSubscribe(自己的Subscription) ↓ MapSubscriber 的内部Subscription 调用 下游的 onSubscribe ↓ LambdaMonoSubscriber 收到 onSubscribe，保存 Subscription ↓ LambdaMonoSubscriber 调用 request(MAX_VALUE) ↓ 请求逐层向上传递，直到源头

请求是“从上往下”传的，数据是“从下往上”发的。这个机制理解了，后面理解背压就很轻松。

3.3 一个完整的调用栈快照

为了让你有更直观的感受，我在一个简单的Mono.just("hello").map(String::length).subscribe(System.out::println)调用处打了个断点，看完整的调用栈，整理成表格：

从第4步到第5步，是整条链路中最重要的“信号传递”。每一步都在建立一对“上下游关系”，直到所有关系的“下游”都拿到Subscription，请求才正式发起。

有些文章喜欢说“订阅是同步的”，这句话对大部分操作符是成立的——只要你没有引入异步调度器（比如subscribeOn、publishOn），整条链路的订阅和请求确实在当前线程内同步完成。

4. 背压协议在Mono中的实现

4.1 Mono的语义：最多一个元素

讨论背压之前，先明确Mono的语义——它最多发出一个元素，然后要么onComplete，要么onError。这跟Flux不同，Flux可能发0个或多个元素，是有持续性的流的抽象。

背压（Backpressure）的本质是：下游告诉上游“我一次能处理多少数据”，上游据此调整发送速率。Mono因为只有最多一个元素，所以它的背压处理天然简单。

源码里可以找到这个设计意图。MonoSubscriber的默认请求量是Long.MAX_VALUE：

class MonoSubscriber<T> implements Subscriber<T>, Subscription, Disposable { T value; @Override public void onSubscribe(Subscription s) { this.s = s; s.request(Long.MAX_VALUE); } }

你可能会想：这不就等同于“不背压”？方向对了一半——Mono的场景里，全量请求是合理的，因为它最多就发一个元素。即使发一个占用内存较大的对象（比如一次数据库查询的结果集），也不会造成无限的资源消耗。

4.2 request(n)的校验逻辑

request(n)在Reactor内部有严格的校验。看Operators.validateRequest：

public static boolean validateRequest(long n) { if (n < 0) { throw new IllegalArgumentException("rule 3.9 - negative request amount"); } return true; }

负数请求是无效的。还有一个隐藏的边界情况是n == 0——如果业务代码显式调用了request(0)，意味着“以后再说”，此时上游不会发数据。

这里有一个实用建议：如果你在自定义Subscriber中控制请求节奏，初始request(1)收到数据后再request(1)，这种“逐条请求”模式在Flux里常见，但在Mono里有些多此一举。因为Mono最多一个元素，一次request(Long.MAX_VALUE)就够了。在Mono里还纠结逐条请求，说明你没吃透Mono的数据规模特性。

4.3 onNext里隐藏的cancel

再看MonoSubscriber处理元素时的关键代码。我第一次看的时候一愣，因为它处理完元素后就直接完成了：

@Override public void onNext(T t) { // 已经终止就不处理 if (STATE.get(this) != TERMINATED) { value = t; // 注意这里，并没有直接触发complete } }

等等，这里有个细节：MonoSubscriber的onNext并不直接调用下游的onComplete。它把值暂存起来，然后通过setValue或后续动作去触发。

那这个“一次性完成”的语义在哪里体现？看Mono.subscribe的最终实现，当你只传了Consumer<? super T>时，内部生成的Subscriber在onNext收到数据后，会调用actual.accept(t)，然后立刻调用cancel()来取消上游的订阅。这个cancel的意义在于：Mono只允许一个元素，既然已经收到了，就没有必要让上游继续生产了。

这个“收到即取消”的细节很关键。如果你自定义实现Subscriber对接Mono，记得一定要在收到唯一元素后主动cancel()，否则上游某些实现了“污染后继续”逻辑的Publisher可能会进一步发更多数据，破坏Mono的单元素协议。

5. 不同Mono变体的subscribe实现差异

5.1 Mono.just：最直接的同步订阅

MonoJust是最简单的Mono实现之一。它的subscribe核心逻辑大概是：

@Override public void subscribe(CoreSubscriber<? super T> actual) { actual.onSubscribe(new ScalarSubscription<>(actual, value)); }

ScalarSubscription是Reactor里一个极度优化的工具类。它实现了一套“正好发送一个值”的机制。request(long n)方法在被调用时，如果之前没有发过元素，就会立刻把value推进actual.onNext(value)，紧接着调用actual.onComplete()。

注意，ScalarSubscription还做了个一致性判断：请求必须大于0才会发数据。这意味着如果你在subscriptionConsumer里设置request(0)然后什么都不做，Mono.just的订阅关系建立了，但数据永远不流动。这不是bug，是背压协议的正确行为。

5.2 Mono.create：异步源的接入

Mono.create算是比较“亲民”的API，很多封装异步回调的代码会用它。它的subscribe调用链多了一层MonoCreate的MonoSink。

MonoSink.success(value)内部做了什么？核心在于它会调用emitter的onNext和onComplete。如果有背压约束，比如下游还没有request，这个数据会被挂起等待吗？

我在实际分析中发现，MonoCreate对“缓存元素”做了处理。MonoSink内部维护一个状态机，如果当前阶段不允许发射，它会暂存本次数据，等到request到达后再补发。这个设计避免了异步场景下“数据先到、请求后到”的时序问题。

这其实是很典型的异步转同步模式。你可以类比成快递柜：快递员（异步回调）到达时发现取快递的人（下游请求）还没来，就把包裹暂存在柜子里，等人来再取。Reactor内部把这个“柜子”用状态机实现，并不复杂但设计得很精妙。

5.3 Mono.defer和subscribeOn：延迟与调度

Mono.defer的subscribe是典型的“每次订阅重新执行函数”逻辑：

@Override public void subscribe(CoreSubscriber<? super T> actual) { Mono<? extends T> p; try { p = supplier.get(); } catch (Throwable e) { // 异常处理 return; } p.subscribe(actual); }

也就是说，Mono.defer本身不产生数据，它只是“懒加载”一个Mono工厂，然后透传订阅。

而Mono.subscribeOn的操作符会在订阅阶段把整个链路的执行切换到指定调度器上。这也是“订阅同步，执行异步”的关键。subscribeOn影响的是订阅和请求的上游传播过程，而不是下游执行——这是很多新手最容易混淆的点。

举个容易踩坑的实际案例：我用Mono.fromCallable(() -> 查询数据库())包了一个耗时的数据库查询，然后在调用处用了subscribeOn(Schedulers.boundedElastic())，但发现执行还是卡主线程。排查后发现subscribeOn只影响了它之前的操作，如果fromCallable写在subscribeOn之前，那个耗时任务确实在弹性线程池执行；但如果我们想的是整个链路的IO操作都切线程，就需要在更高层级统一切换。理解“位置决定影响范围”这个要点，能帮你少走很多弯路。

6. 重载方法的细节与陷阱

6.1 三参/四参subscribe的参数组合

Reactor为subscribe提供了一组重载方法，从无参到四参，每个多出来的参数对应一个“更细粒度的回调”。我遇到很多开发者在业务代码中只使用最基础的无参版本，但排查问题时才发现自己缺少了对特定事件的感知能力。

四参数版本最不常用，但不代表它不重要。我个人的经验是：如果你要在订阅关系刚建立时做一些初始化操作（比如记录开始时间、注册取消回调），subscriptionConsumer是标准的扩展点。注意，在这个回调里，你可以拿到Subscription，也就可以调用request(0)来“暂停”数据流动，这在需要先做一些前置异步校验的场景中很有效。

6.2 空值校验与陷阱

看这个细节：在Mono的subscribe实现中，对errorConsumer和completeConsumer的处理是分开的。源码逻辑大致是：

if (actual != null) { // 包装 actual } if (errorConsumer != null) { // 转换成 onError 回调 } else { // 默认行为：抛异常 }

也就是说：actual为null是允许的，此时订阅链只会“完成”不会“发数据”。completeConsumer为null也是允许的，完成信号被丢弃。唯独errorConsumer为null时，异常会被重新抛出。

这在第三方库集成中挺常见的——有的库用subscribe(consumer, null, runnable)只关心成功与完成，但一旦出错，异常会直接冒泡。这不是错误，是故意设计：宁可让异常可见，也不要静默吞掉。

6.3 doOn系列与subscribe的配合时机

很多人分不清doOnNext和subscribe里的Consumer执行的先后。其实逻辑很简单：

• doOnNext、doOnError、doOnComplete这些方法是在数据流经它们所在位置时触发，位置决定时机。
• subscribe里的回调是整个链路的终点，一定是最后触发。

举个例子：

Mono.just("A") .doOnNext(v -> log.info("doOnNext: {}", v)) .doOnComplete(() -> log.info("doOnComplete")) .subscribe(v -> log.info("subscribe: {}", v));

输出顺序是：doOnNext: A、doOnComplete、subscribe: A。为什么subscribe的Consumer反而在doOnComplete之后？因为LambdaMonoSubscriber的内部逻辑在接收到元素并调用actual.accept(t)之后，会立刻触发完成状态。也就是说，“元素处理”和“完成信号”是紧挨着的，甚至可以在同一次栈周期内完成。理解这个顺序对排查日志时间线混乱的问题很有帮助。

7. 实操总结：源码分析后最值得带走的几件事

7.1 从入口函数和调用栈开始读源码

我个人读Reactors源码的经验是：不要一上来就去翻Mono这个超大类，而应从一个具体的操作符（比如Mono.map）的subscribe方法入手，打上断点，跑一个最简单的Demo，观察完整的调用栈。一旦你理解了“订阅请求反向传播、数据正向传递”这个核心脉络，其他操作符的源码就是在此基础上添砖加瓦。

我自己带团队时经常给同学们安排一个练习：让每个人都把Mono.just(1).map(x -> x + 1).subscribe(System.out::println)的调用栈画出来，画不出来就去看源码。这个练习做完，再聊“响应式为什么是懒的”，大家基本都能有自己的理解了。

7.2 一个容易忽略的资源释放问题

最后分享一个真实案例。之前有个同事在业务里用了Mono.create封装一个长连接消息监听，把MonoSink保存到了成员变量里，想着之后通过sink.success(data)推送数据。结果他的同事在外部直接调用了dispose()，之后MonoSink的状态已变成TERMINATED。此时再调用success，会发现数据根本推不出去，也没有任何异常提示。

看了MonoSink的源码，你会发现每个success/error调用都会经过状态检查。如果当前状态已经是TERMINATED，调用会被静默忽略。这个“静默丢弃”的设计保护了响应式流的协议一致性，但也给业务埋了一个不大不小的坑：如果你依赖sink推送数据，一定要处理好sink的生命周期，或者在下游加个doFinally做清理逻辑，避免状态混乱。

7.3 学到的不仅是源码，更是一种设计范式

看完Mono的subscribe实现，我最大的收获其实不是“我知道了某个方法怎么调用”，而是理解了Reactor内部状态机的设计方式。它用原子状态位、CAS操作、回调的不可重入等组合方式，优雅地解决了异步并发中最难处理的竞态问题。

以后遇到类似的需求——比如自己设计一个带取消能力的异步任务系统，完全可以复用这套思路：一个AtomicInteger做状态标识，所有回调执行前先CAS检查状态，终态后任何触点都拦截。这套设计模式放之四海而皆准。

Reactors的代码量很大，但最有价值的部分恰恰是这些“根节点”上的基础方法。吃透Mono的subscribe，后面再去啃Flux、背压操作符、自定义操作算子，都有了稳固的地基。源码分析这种事，慢就是快，把一个入口吃透，胜过草草扫过十个类。