ARTICLE · 1213411

Reactor 源码分析

发布时间：2026-10-20 01:47:13 最近访问：2026-10-20 01:47:15

Reactor 源码分析

文档类型：技术源码分析文档适用人群：Java 后端、响应式编程开发人员依赖版本：reactor-core: 3.6.x 简述：本文介绍 Reactor 核心概念、底层核心设计，附带简易源码解读，同时提供多种 Mock 测试示例（单元测试场景）

1. Reactor 简介

Reactor 是 JVM 上的响应式编程实现，是 Spring WebFlux 的底层核心库，实现了 Reactive Streams 规范。响应式编程核心思想：事件驱动、异步非阻塞、数据流推送、支持背压。

对比传统命令式：命令式：主动拉取数据（get()），等待返回，阻塞线程响应式：数据就绪后推送给订阅者，订阅者可以控制请求多少数据（背压）

Reactor 提供两个核心类型：

Mono<T>
：0 或 1 个元素的异步数据流（单个结果、空、异常）
Flux<T>
：0 ~ N 个元素的异步数据流（流式列表）

核心包：reactor-coreReactive Streams 4 个核心接口规范：Publisher, Subscriber, Subscription, Processor

2. 核心概念：Publisher / Subscriber / Subscription

2.1 Publisher

public interface Publisher<T> {    // 订阅：订阅者注册到发布者    voidsubscribe(Subscriber<? super T> s);}

Flux / Mono 本质就是 Publisher。调用 subscribe() 才会触发数据流（冷流：只有订阅才开始执行）。

冷流 (Cold Stream)：每次 subscribe 都会重新执行一遍源逻辑；大多数 Reactor 操作符都是冷流。

2.2 Subscriber

public interface Subscriber<T> {    // 订阅成功回调，传入Subscription，用于控制流    voidonSubscribe(Subscription s);    // 收到数据元素    voidonNext(T t);    // 流正常结束    voidonComplete();    // 发生异常终止流    voidonError(Throwable t);}

2.3 Subscription

public interface Subscription {    // 请求n条数据（背压核心）    voidrequest(long n);    // 取消订阅，终止数据流    voidcancel();}

重点：request(n) 是背压的核心，订阅者告诉上游：我现在能处理多少条数据，上游不会超过这个数量推送。

生命周期

subscribe() → onSubscribe → request(N) → onNext → (onComplete | onError)

3. Flux & Mono 源码核心逻辑分析

3.1 Flux 基础抽象

Flux 是抽象类，所有操作符返回的都是 Flux 的子类（如FluxJust、FluxMap、FluxFlatMap）。

源码关键点：所有操作符只是组装描述，不执行逻辑，直到 subscribe。

示例：Flux.just(1,2,3).map(i -> i*2)

Flux.just创建 FluxJust，代表固定数据源
.map()创建 FluxMap，持有上游 Flux 和 map 转换函数
这两行只是构建链路，没有执行任何计算
调用 .subscribe() 才会从下游向上建立订阅链，开始推送数据

简化伪源码（FluxMap）

// 简化版 FluxMap 源码示意final class FluxMap<T, R> extends Flux<R> {    private final Flux<T> source;    private final Function<T, R> mapper;    public FluxMap(Flux<T> source, Function<T, R> mapper) {        this.source = source;        this.mapper = mapper;    }    @Override    public void subscribe(Subscriber<? super R> actual) {        // 包装下游订阅者，构建中间订阅器        source.subscribe(new MapSubscriber<>(actual, mapper));    }    static final class MapSubscriber<T,R> implements Subscriber<T>, Subscription {        final Subscriber<? super R> actual;        final Function<T,R> mapper;        Subscription upstreamSubscription;        public MapSubscriber(Subscriber<? super R> actual, Function<T, R> mapper) {            this.actual = actual;            this.mapper = mapper;        }        @Override        public void onSubscribe(Subscription s) {            this.upstreamSubscription = s;            // 把订阅传给下游            actual.onSubscribe(this);        }        @Override        public void onNext(T t) {            R value = mapper.apply(t);            actual.onNext(value);        }        @Override        public void onComplete() {            actual.onComplete();        }        @Override        public void onError(Throwable throwable) {            actual.onError(throwable);        }        @Override        public void request(long n) {            upstreamSubscription.request(n);        }        @Override        public void cancel() {            upstreamSubscription.cancel();        }    }}

核心：每一个操作符都包装一层 Subscriber，形成订阅链，事件从上游逐层向下传递；request/cancel 从下游向上传递。

3.2 Mono

Mono 同样继承 Publisher，逻辑和 Flux 一致，约束最多推送 1 个 onNext。如果 Mono 收到多个 onNext，会抛出异常，这是 Mono 的校验逻辑。

4. 背压 (Backpressure) 机制原理

Reactive Streams 规范强制要求背压：下游可以控制上游发送速率，防止 OOM。

下游通过 request(long n) 声明可处理元素数量
上游不能推送超过 request 数量的数据
当下游处理慢，上游会暂停推送，而不是把数据放到无限队列

Reactor 内置策略：
onBackpressureBuffer
：缓存溢出数据
onBackpressureDrop
：丢弃来不及处理的数据
onBackpressureError
：溢出直接抛异常

注意：很多同步源（Flux.just, Flux.fromIterable）在背压不足时会直接在 onNext 抛出IllegalStateException，因为同步源无法限速。

5. Reactor Mock 测试方案

依赖（JUnit5 + reactor-test）<!-- pom.xml --><dependencies>    <dependency>        <groupId>io.projectreactor</groupId>        <artifactId>reactor-core</artifactId>        <version>3.6.4</version>    </dependency>    <dependency>        <groupId>io.projectreactor</groupId>        <artifactId>reactor-test</artifactId>        <version>3.6.4</version>        <scope>test</scope>    </dependency>    <dependency>        <groupId>org.junit.jupiter</groupId>        <artifactId>junit-jupiter-api</artifactId>        <version>5.9.2</version>        <scope>test</scope>    </dependency>    <dependency>        <groupId>org.junit.jupiter</groupId>        <artifactId>junit-jupiter-engine</artifactId>        <version>5.9.2</version>        <scope>test</scope>    </dependency></dependencies>

Reactor 测试两大工具：
StepVerifier
：官方推荐，验证 Flux/Mono 数据流（最常用）
TestPublisher
：用来 Mock 上游 Publisher，主动发射数据、异常、完成信号

区分：
StepVerifier
：消费端，校验输出
TestPublisher
：生产端，Mock 数据源

6. Mock 多种使用示例

测试场景：假设业务服务 UserService 返回 Mono<User> / Flux<Order>

示例 1：TestPublisher Mock Mono 正常返回

import org.junit.jupiter.api.Test;import reactor.core.publisher.Mono;import reactor.test.publisher.TestPublisher;import reactor.test.StepVerifier;public class ReactorMockDemoTest {    @Test    void testMockMonoSuccess() {        // 构造Mock发布者        TestPublisher<String> mockPublisher = TestPublisher.create();        Mono<String> mono = mockPublisher.mono();        // 业务处理链路（模拟service转换逻辑）        Mono<String> processMono = mono.map(s -> "Hello:" + s);        // StepVerifier订阅        StepVerifier.create(processMono)                .then(() -> {                    // 手动发射数据                    mockPublisher.emit("reactor");                })                .expectNext("Hello:reactor")                .verifyComplete();    }}

示例 2：TestPublisher Mock 异常场景（onError）

@Testvoid testMockMonoError() {    TestPublisher<String> mockPublisher = TestPublisher.create();    Mono<String> mono = mockPublisher.mono();    StepVerifier.create(mono)            .then(() -> mockPublisher.error(new RuntimeException("service timeout")))            .expectError(RuntimeException.class)            .verify();}

示例 3：TestPublisher Mock Flux 流式多元素

@Testvoid testMockFlux() {    TestPublisher<Integer> mockFluxPublisher = TestPublisher.create();    var flux = mockFluxPublisher.flux()            .filter(i -> i > 0);    StepVerifier.create(flux)            .then(() -> mockFluxPublisher.next(1, 2, 3).complete())            .expectNext(1,2,3)            .verifyComplete();}

示例 4：结合 Mockito Mock Service + StepVerifier（业务最常用）

场景：Service 层返回 Mono，使用 Mockito mock service 对象，配合 StepVerifier 校验结果

import org.junit.jupiter.api.Test;import org.junit.jupiter.api.extension.ExtendWith;import org.mockito.InjectMocks;import org.mockito.Mock;import org.mockito.junit.jupiter.MockitoExtension;import reactor.core.publisher.Mono;import reactor.test.StepVerifier;import static org.mockito.Mockito.when;@ExtendWith(MockitoExtension.class)public class UserServiceTest {    @Mock    private UserRepository userRepository;    @InjectMocks    private UserService userService;    @Test    void testGetUserById() {        User mockUser = new User(1L, "zhangsan");        // Mock repository 返回Mono        when(userRepository.findById(1L)).thenReturn(Mono.just(mockUser));        Mono<User> result = userService.getById(1L);        StepVerifier.create(result)                .expectNextMatches(user -> user.getId().equals(1L))                .verifyComplete();    }    // 实体类    static class User {        private Long id;        private String name;        public User(Long id, String name) {            this.id = id;            this.name = name;        }        public Long getId() { return id; }    }}

示例 5：Mock 空流场景（Mono.empty）

@Testvoid testMonoEmpty() {    TestPublisher<String> publisher = TestPublisher.create();    Mono<String> mono = publisher.mono();    StepVerifier.create(mono)            .then(publisher::complete) // 直接完成，无数据            .verifyComplete();}

示例 6：背压场景的 Mock（控制 request 数量）

@Testvoid testBackpressureMock() {    TestPublisher<Integer> testPublisher = TestPublisher.create();    var flux = testPublisher.flux();    StepVerifier.create(flux, 2) // 初始request=2            .then(() -> testPublisher.next(1,2,3,4))            .expectNext(1,2)            .thenRequest(2) // 再请求2个            .expectNext(3,4)            .verifyComplete();}

重点说明：
TestPublisher
适合模拟上游数据源，可以精细控制什么时候发 next、complete、error
StepVerifier
是断言工具，验证数据流输出、异常、完成信号、时序
项目开发中一般两种组合使用：Mockito mock 业务组件 + StepVerifier 校验响应式返回；底层数据源测试用 TestPublisher

7. Reactor 线程模型（Schedulers）源码解读

7.1 Scheduler 核心定位

Reactor 所有异步调度由 Scheduler 接口负责，本质是响应式版本的线程池，用来切换执行线程。

关键：默认情况下，Reactor 操作符不会自动切换线程，在哪个线程调用 subscribe，就在哪个线程执行（同步执行）。只有显式使用 publishOn / subscribeOn 才会切换线程。

核心接口定义简化：

public interface Scheduler {    // 创建可调度任务    Worker createWorker();    // 调度任务延迟执行    Disposable schedule(Runnable task);    Disposable schedule(Runnable task, long delay, TimeUnit unit);    interface Worker extends Disposable {        Disposable schedule(Runnable task);    }}

7.2 内置 Scheduler 类型

// 1. immediate：当前线程直接执行（默认）Schedulers.immediate()// 2. single：单一线程，顺序执行任务Schedulers.single()// 3. elastic：弹性线程池（旧版，3.4+推荐使用boundedElastic）Schedulers.boundedElastic()// 4. parallel：固定数量线程池，CPU密集场景，线程数默认=CPU核心数Schedulers.parallel()

7.3 publishOn 与 subscribeOn 源码逻辑（重点）

subscribeOn
：影响上游整条链路的订阅建立阶段，决定 subscribe 动作在哪个线程执行，只生效一次，写在任何位置效果一致。
publishOn
：只影响下游 onNext/onError/onComplete 事件的执行线程，写在哪个位置，就从该位置往后切换线程；可以多次调用，多次切换线程。

简化伪源码 FluxPublishOn：

final class FluxPublishOn<T> extends Flux<T> {    private final Flux<T> source;    private final Scheduler scheduler;    @Override    public void subscribe(Subscriber<? super T> actual) {        Scheduler.Worker worker = scheduler.createWorker();        source.subscribe(new PublishOnSubscriber<>(actual, worker));    }    static class PublishOnSubscriber<T> implements Subscriber<T> {        private final Subscriber<? super T> downstream;        private final Scheduler.Worker worker;        @Override        public void onNext(T value) {            // 把onNext事件交给scheduler线程执行            worker.schedule(() -> downstream.onNext(value));        }    }}

7.4 线程调度 Mock 测试注意点

异步调度会带来线程等待问题，单元测试推荐：

测试代码中，使用 Schedulers.immediate() 替换业务里的 boundedElastic，消除多线程时序不稳定
StepVerifier 内部自带虚拟时间调度器 withVirtualTime()，适合测试延迟、定时任务，无需 sleep

// 虚拟时间示例，不需要真实等待10s

@Testvoid testDelayWithVirtualTime() {    Flux<Long> flux = Flux.just(1L)            .delayElements(Duration.ofSeconds(10));    StepVerifier.withVirtualTime(() -> flux)            .expectSubscription()            .thenAwait(Duration.ofSeconds(10))            .expectNext(1L)            .verifyComplete();}

业务选型建议
IO 阻塞（DB、RPC 调用）：boundedElastic
CPU 计算：parallel
单元测试：虚拟时间 + immediate

8. Reactor 热流 ConnectableFlux 源码与 Mock 测试

8.1 冷流 vs 热流

冷流（Cold）
：subscribe 才开始生产数据，每个订阅者独立数据流，如Flux.just, Flux.fromIterable
热流（Hot）
：不管有没有订阅者，数据源持续产生数据；多个订阅者共享同一份数据流。ConnectableFlux 是 Reactor 经典的可控制启动的热流，必须手动调用connect()之后，上游才开始发射数据。

8.2 ConnectableFlux 核心源码简化

// publish() 返回 ConnectableFluxFlux<Integer> source = Flux.interval(Duration.ofMillis(100))        .take(5)        .map(Long::intValue)        .publish(); // 转成ConnectableFlux// publish底层：FluxPublish，内部保存所有订阅者，connect触发上游订阅public class FluxPublish<T> extends ConnectableFlux<T> {    private final Flux<T> source;    List<Subscriber<? super T>> subscribers;    // 多个subscriber订阅时，先存入列表，不会立刻订阅source    @Override    public void subscribe(Subscriber<? super T> s) {        subscribers.add(s);    }    // 调用connect，才去订阅上游source，开始推送数据给所有订阅者    @Override    public void connect(Consumer<? super Disposable> cancelSupport) {        source.subscribe(new PublishCentralSubscriber());    }}

常用 API
.publish()
→ ConnectableFlux
.autoConnect(n)
：自动 connect，订阅数达到 n 个自动启动
.refCount(n)
：引用计数，订阅达到 n 个启动，全部取消订阅自动断开上游

8.3 ConnectableFlux Mock 测试示例

场景：多订阅者共享同一个热流，调用 connect 之后才发射数据

@Testvoid testConnectableFluxMock() {    // Mock上游数据源    TestPublisher<String> testPublisher = TestPublisher.create();    ConnectableFlux<String> connectableFlux = testPublisher.flux().publish();    // 订阅者A    var verifierA = StepVerifier.create(connectableFlux)            .expectNext("msg1", "msg2")            .verifyLater();    // 订阅者B    var verifierB = StepVerifier.create(connectableFlux)            .expectNext("msg1", "msg2")            .verifyLater();    // 此时还没connect，不会发射任何数据    // 手动connect，启动上游流    Disposable disposable = connectableFlux.connect();    // 发射数据，两个订阅者同时收到消息    testPublisher.next("msg1", "msg2").complete();    verifierA.verifyComplete();    verifierB.verifyComplete();    disposable.dispose();}

8.4 热流测试要点

connect 调用时机决定数据开始发送；connect 前产生的数据会丢失（无订阅者接收）
refCount 自动管理连接，适合消息广播场景
TestPublisher 非常适合模拟热流数据源，控制消息发送时机，验证多订阅者广播行为

9. 开发踩坑常见问题排查

9.1 block () 滥用问题

block() 会阻塞当前线程，生产代码禁止使用。

问题：WebFlux 中 block 会占用事件循环线程，引发服务卡死
解决：全部使用响应式链式调用，单元测试用 StepVerifier 代替 block

9.2 冷流重复订阅，多次执行上游逻辑

Mono<String> mono = Mono.fromCallable(() -> {    System.out.println("执行RPC");    return "result";});mono.subscribe();mono.subscribe(); // 会再次执行RPC！

解决方案：使用 .cache() 缓存结果，转成热流。

9.3 背压溢出异常

同步数据源（Flux.just）不支持限流，当下游 request 不足会抛异常。解决方案：onBackpressureXXX 系列操作符做溢出策略。

9.4 publishOn /subscribeOn 混淆

subscribeOn 只改变上游订阅建立线程；publishOn 改变下游事件执行线程，写的位置非常关键。

9.5 异常没有被捕获，流直接终止

Mono/Flux 出现 onError 之后流直接结束；需要使用 onErrorResume / onErrorReturn 捕获异常，保障链路不中断。

10. 总结

Reactor 是实现 Reactive Streams 的响应式库，核心是Mono/Flux，所有操作符懒加载，subscribe 才执行。
数据流是链式 Subscriber，事件向下传递，背压 request 信号向上传递。
Mock 方案有两条路线：

业务层：Mockito mock Bean，返回构造好的 Mono/Flux，配合StepVerifier校验结果
底层数据源：TestPublisher 手动控制发射事件，适合测试背压、时序、异常边界

Scheduler 是 Reactor 线程调度核心；publishOn控制下游事件线程，subscribeOn控制订阅线程。
ConnectableFlux 是可控热流，多订阅者共享数据流，常用于消息广播场景；测试时需要注意 connect 时机。
单元测试最佳实践：不要阻塞获取 block ()，全部使用 StepVerifier，避免测试死锁与时序问题；异步场景优先使用虚拟时间消除 sleep。