Project Reactor作为响应式编程范式的核心实现框架,严格遵循Reactive Streams规范体系,其架构设计完整包含了规范定义的四个核心组件:Publisher(数据源)、Subscriber(订阅者)、Subscription(订阅关系)和Processor(处理节点)。在该框架中,Flux和Mono不仅实现了Publisher接口的标准语义,更构建了完整的响应式数据流处理范式:通过订阅关系建立生产-消费通道,基于事件驱动机制实现非阻塞式数据推送,同时通过背压(backpressure)协议保障系统的弹性通信。
从整体上理解 Project Reactor 的工作原理,能够帮助我们更清晰地掌握其中的各种概念和操作,避免迷失方向。实际上,从大局来看,整个 Reactor 就是基于订阅-发布模式的。Flux 和 Mono 作为系统中默认的 Publisher,简化了我们自定义 Publisher 的工作。Flux 和 Mono 集成了大量的操作符,这些操作符的存在减少了我们自定义 Subscriber 和 Processor 的需求。通过这些操作符的组合,我们可以直接对数据源和元素进行操作,而无需自己编写额外的 Processor 和 Subscriber。除非在特殊情况下,否则不建议主动去自定义 Subscriber 和 Processor。
创建数据源(Flux,Mono)->转换和处理数据(map,filter...)->subscribe订阅数据源
作为Project Reactor的核心发布者,Flux和Mono的主要区别如下:
Flux代表0-N个元素的异步序列
Mono表示0-1个结果的异步操作
// 创建Flux
Flux.just("1", "2", "3").subscribe(System.out::println);
// 创建Mono
Mono.just("a").subscribe(System.out::println);
了解了Flux和Mono之后,我们知道了如何简单的创建数据源,其中Flux和Mono也给我们提供了非常多的创建数据源的方式,大概分为以下几类。
Mono.create
和 Flux.generate
/Flux.create
允许手动控制元素发射(同步或异步)。Future
、Callable
或 Supplier
中获取数据,支持非阻塞操作。Mono.delay
延迟发射,Flux.interval
周期性发射递增数值。zip
严格对齐元素,merge
无序合并,concat
顺序连接。FluxSink
或 MonoSink
手动控制背压和元素发射。Mono 和 Flux 数据源创建方式分类总结
类别 | 描述 | Mono 方法示例 | Flux 方法示例 |
---|---|---|---|
空数据源 | 创建不发射任何元素的数据流。 | Mono.empty() |
Flux.empty() |
单个元素 | 发射单个静态值或对象。 | Mono.just(T) |
Flux.just(T...) |
多个元素 | 发射多个静态值或对象(仅 Flux 支持)。 |
N/A |
Flux.just(T1, T2...) |
集合/数组 | 从集合或数组生成元素。 | N/A |
Flux.fromIterable(List Flux.fromArray(T[]) |
流(Stream) | 从 Java Stream 生成元素。 |
N/A |
Flux.fromStream(Stream |
动态生成 | 通过生成器函数动态生成元素。 | Mono.create(sink -> {...}) |
Flux.generate(sink -> {...}) Flux.create(sink -> {...}) |
异步数据源 | 从异步操作(如 Future 、Callable )获取数据。 |
Mono.fromFuture(Future) Mono.fromCallable(Callable) |
Flux.from(Publisher) Flux.fromStream(Supplier |
错误信号 | 直接发射错误信号。 | Mono.error(Throwable) |
Flux.error(Throwable) |
延迟初始化 | 惰性生成数据(订阅时才执行逻辑)。 | Mono.defer(() -> ...) Mono.fromSupplier(Supplier) |
Flux.defer(() -> ...) Flux.fromStream(Supplier |
时间驱动 | 基于时间生成数据(如定时、延迟)。 | Mono.delay(Duration) |
Flux.interval(Duration) |
合并/组合 | 合并多个数据源。 | Mono.zip(Mono1, Mono2...) |
Flux.merge(Flux1, Flux2...) Flux.concat(Flux1, Flux2...) Flux.zip(Flux1, Flux2...) |
背压适配 | 适配外部背压机制(如 Sink 手动控制)。 |
Mono.create(MonoSink) |
Flux.create(FluxSink) |
条件触发 | 根据条件生成数据(如 first 、takeUntil )。 |
Mono.firstWithValue(Mono1, Mono2) |
Flux.firstWithValue(Publisher...) Flux.takeUntil(Predicate) |
Project Reactor 的发布模型是其响应式编程的核心机制,主要分为 冷发布者(Cold Publisher) 和 热发布者(Hot Publisher)。它们的区别在于数据流的生成、共享方式以及订阅者的消费行为。以下是详细解释:
定义:冷发布者为每个订阅者生成独立的数据流。每个订阅者都会触发数据源的完整生成过程,即使其他订阅者已订阅过。
特点:
适用场景:
需要每个订阅者获取完整数据(如 HTTP 请求、数据库查询)。
数据源的生成成本较高,但需确保订阅者的独立性。
代码示例
// 创建冷发布者
Flux coldFlux = Flux.range(1, 3).doOnNext(i -> System.out.println("冷发布者发出: " + i));
// 第一个订阅者
coldFlux.subscribe(i -> System.out.println("订阅者1: " + i));
// 第二个订阅者
coldFlux.subscribe(i -> System.out.println("订阅者2: " + i));
输出
冷发布者发出: 1
订阅者1: 1
冷发布者发出: 2
订阅者1: 2
冷发布者发出: 1
订阅者2: 1
冷发布者发出: 2
订阅者2: 2
定义:热发布者共享一个统一的数据流,所有订阅者消费同一份数据。数据源的生成与订阅者的订阅时间无关,后订阅的订阅者可能错过早期数据。
特点:
适用场景:
热发布者的实现方式有如下几种:
ConnectableFlux(手动控制)
通过 publish()
方法将 Flux
转换为 ConnectableFlux
,需手动调用 connect()
启动数据流。
代码示例
// 创建 ConnectableFlux 并转换为热发布者
ConnectableFlux hotFlux = Flux.range(1, 3)
.doOnNext(i -> System.out.println("热发布者发出: " + i))
.publish(); // 转换为 ConnectableFlux
// 订阅者A
hotFlux.subscribe(i -> System.out.println("订阅者A: " + i));
// 订阅者B
hotFlux.subscribe(i -> System.out.println("订阅者B: " + i));
// 手动触发数据流开始
hotFlux.connect();
输出
热发布者发出: 1
订阅者A: 1
订阅者B: 1
热发布者发出: 2
订阅者A: 2
订阅者B: 2
热发布者发出: 3
订阅者A: 3
订阅者B: 3
autoConnect()
(自动连接)当达到指定订阅者数量时,自动启动数据流。
Flux autoFlux = Flux.range(1, 3)
.doOnNext(i -> System.out.println("热发布者发出: " + i))
.publish()
.autoConnect(2);// 当有 2 个订阅者时自动启动
autoFlux.subscribe(i -> System.out.println("订阅者A: " + i));
autoFlux.subscribe(i -> System.out.println("订阅者B: " + i));
输出
热发布者发出: 1
订阅者A: 1
订阅者B: 1
热发布者发出: 2
订阅者A: 2
订阅者B: 2
热发布者发出: 3
订阅者A: 3
订阅者B: 3
share()
(简化热发布者)等价于 publish().refCount(1)
:当第一个订阅者到来时启动,最后一个取消订阅时终止。
Flux sharedFlux = Flux.interval(Duration.ofSeconds(1))
.doOnNext(i -> System.out.println("热发布者发出: " + i))
.take(5)
.share();
sharedFlux.subscribe(i -> System.out.println("订阅者A: " + i));
Thread.sleep(2500);
sharedFlux.subscribe(i -> System.out.println("订阅者B: " + i)); // 订阅者B错过前2个数据
输出
热发布者发出: 0
订阅者A: 0
热发布者发出: 1
订阅者A: 1
热发布者发出: 2
订阅者A: 2
订阅者B: 2
热发布者发出: 3
订阅者A: 3
订阅者B: 3
热发布者发出: 4
订阅者A: 4
订阅者B: 4
replay()
(历史数据缓存)允许新订阅者消费订阅前的历史数据(缓存策略可配置)。
ConnectableFlux replayFlux = Flux.range(1, 3)
.doOnNext(i -> System.out.println("热发布者发出: " + i))
.replay(2);// 缓存最近2个数据
replayFlux.subscribe(i -> System.out.println("订阅者A: " + i));
replayFlux.connect();
Thread.sleep(1000);
replayFlux.subscribe(i -> System.out.println("订阅者B: " + i)); // 订阅者B收到最后2个数据
输出:
热发布者发出: 1
订阅者A: 1
热发布者发出: 2
订阅者A: 2
热发布者发出: 3
订阅者A: 3
订阅者B: 2
订阅者B: 3
冷发布者和热发布者对比表格
特性 | 冷发布者 | 热发布者 |
---|---|---|
数据生成时机 | 订阅时生成 | 提前生成(或由 connect() 触发) |
订阅者独立性 | 每个订阅者独立消费完整数据 | 共享同一数据流 |
资源消耗 | 高(每个订阅者独立生成) | 低(共享生成逻辑) |
典型场景 | 数据库查询、静态数据 | 实时事件、广播 |
类别 | 操作符示例 | 功能描述 |
---|---|---|
转换操作符 | buffer , map , flatMap , window |
修改流中元素结构或内容(如分组、映射、扁平化) |
过滤操作符 | filter , take , skip |
按条件筛选元素(如保留满足条件的元素、跳过前N项) |
组合操作符 | merge , concat , zip |
合并多个流(如按顺序连接、并行合并、元素一一配对) |
条件操作符 | any , all , hasElement |
判断流中元素是否满足条件(如是否存在满足条件的元素) |
数学操作符 | count , sum , reduce |
对元素进行聚合计算(如统计总数、求和、累加) |
错误处理操作符 | onErrorReturn , onErrorResume |
异常时提供备选值或切换至备用流(如返回静态值、动态恢复逻辑) |
工具操作符 | delay , timeout , log , subscribe |
控制流生命周期(如延迟发送、超时中断、记录日志、触发订阅) |
整个数据源操作 | doOnNext ,, ,doOnRequest ,doOnSubscribe ,doOnComplete 等 |
其中以doOn开头的可以对整个数据链的不同状态进行操作 |
//转换操作符、过滤操、条件及数学操作符类似Java的Stream这里不做过多赘述
//map
Flux.just(1, 2, 3).map(i -> i + 1).subscribe(System.out::println);
//filter
Flux.just("a", "b", "c").filter(s -> s.equals("a")).subscribe(System.out::println);
//flatMap
Flux.just("a", "b", "c").flatMap(s -> Flux.just(s.toUpperCase())).subscribe(System.out::println);
//reduce
Flux.just(1, 2, 3).reduce(0, (a, b) -> a + b).subscribe(System.out::println);
//window 窗口使用
Flux.just(1, 2, 3, 4, 5, 6).window(3, 1).flatMap(e -> e.reduce(0, Integer::sum)).subscribe(System.out::println);
//buffer 背压或者批处理使用,会缓存数据
Flux.just(1, 2, 3, 4, 5, 6).buffer(3, 1).subscribe(System.out::println);
zip操作符可以将多个(最多8个)流合并成一个流,合并的方式是将两个流中的元素按照顺序一一对应,然后将两个元素组合成一个元素。 如果两个流的长度不一致,那么最终合并成的流的长度就是两个流中长度较短的那个流的长度。
Flux flux1 = Flux.just("a", "b", "c");
Flux flux2 = Flux.just("d", "e", "f");
Flux flux3 = Flux.just("1", "2", "3");
Flux.zip(flux1, flux2, flux3).subscribe(System.out::println);
//输出
[a,d,1]
[b,e,2]
[c,f,3]
merge 操作符可以将两个流合并成一个流,合并的方式是将两个流中的元素交替地放入到合并后的流中。同时运行,根据时间先后运行。
Flux flux3 = Flux.just(1, 2, 3).delayElements(Duration.ofMillis(80));
Flux flux4 = Flux.just(4, 5, 6).delayElements(Duration.ofMillis(50));
flux3.mergeWith(flux4).subscribe(System.out::println);
//输出 由于是根据时间先后处理,所以结果大概率是这样,也有可能会稍有不同
4
1
5
2
6
3
concat 操作符可以将两个流合并成一个流,合并的方式是将两个流中的元素按照顺序放入到合并后的流中。按照顺序分别运行,flux1运行完成以后再运行flux2
Flux flux1 = Flux.just(1, 2, 3).delayElements(Duration.ofMillis(80));
Flux flux2 = Flux.just(4, 5, 6).delayElements(Duration.ofMillis(50));
flux1.concatWith(flux2).subscribe(System.out::println);
//输出
1
2
3
4
5
6
Project Reactor 提供了大量的以doOn开头的方法,这些方法用于在数据流的生命周期中插入副作用逻辑(如日志、监控或资源管理),不修改数据流本身,仅用于观察或触发行为。
每个方法的使用方法大致相同,下面以doOnRequest和doOnNext做一下简单的示例。
Flux.just(1, 2, 3, 4, 5, 6).doOnNext(s -> System.out.println("doOnNext: " + s)).subscribe();
System.out.println("----------------");
Flux.just(1, 2, 3).doOnRequest(s -> System.out.println("doOnRequest: " + s)).subscribe(System.out::println);
//输出
doOnNext: 1
doOnNext: 2
doOnNext: 3
doOnNext: 4
doOnNext: 5
doOnNext: 6
----------------
doOnRequest: 9223372036854775807
1
2
3
下面是每个方法的使用场景和触发时机。
方法 | 触发时机 | 参数类型 | 典型场景 |
---|---|---|---|
doOnSubscribe |
订阅时 | Consumer |
资源初始化 |
doOnNext |
元素推送时 | Consumer |
日志记录、状态更新 |
doOnError |
发生错误时 | Consumer |
错误监控、报警 |
doOnComplete |
流正常结束时 | Runnable |
完成通知 |
doOnRequest |
下游请求数据时 | Consumer |
背压调试、请求量监控 |
doOnCancel |
取消订阅时 | Runnable |
资源释放 |
doOnEach |
所有事件发生时 | Consumer |
统一事件处理 |
doOnTerminate |
流终止前(完成/错误前) | Runnable |
终止前清理逻辑 |
doAfterTerminate |
流终止后(完成/错误后) | Runnable |
终止后统计 |
doOnDiscard |
元素被丢弃时 | Consumer |
资源回收、数据一致性检查 |
subscribe 操作符用来订阅流中的元素。 当流中的元素没有被订阅的时候,所有的操作都不会触发,只有当流中的元素被订阅的时候,所有的操作才会触发。 通过上面内容的阅读,相信你已经对Project Reactor的发布订阅模型已经了解了个大概,上面的订阅的例子也有很多,这里不做过多的赘述。
Schedulers
是管理线程和并发任务的核心工具,用于控制响应式流的执行上下文。通过合理选择调度器,可以优化资源利用、避免阻塞,并提升应用性能
调度器 | 线程模型 | 适用场景 | 注意事项 |
---|---|---|---|
immediate |
当前线程 | 轻量级同步操作 | 避免阻塞 |
single |
单线程 | 严格顺序执行 | 避免长时间阻塞 |
boundedElastic |
动态线程池 | 阻塞 I/O 操作 | 控制最大线程数和队列容量 |
parallel |
固定大小线程池 | 计算密集型并行任务 | 线程数默认等于 CPU 核心数 |
fromExecutorService |
自定义线程池 | 集成现有线程池 | 需自行管理生命周期 |
在 Project Reactor 中,可以很方便的通过publishOn和subscribeOn来切换使用的线程调度器。
Flux.range(1, 10)
.publishOn(Schedulers.boundedElastic()) //切换调度器
.log("publish thread:")
.flatMap(n -> Mono.fromCallable(() -> n).subscribeOn(Schedulers.parallel())) //切换调度器
.log("subscribe thread:")
.subscribe();
当然在JDK17及更改的版本中也可以结合虚拟线程进一步提高并发量。
Scheduler customSchedule = Schedulers.fromExecutor(Executors.newVirtualThreadPerTaskExecutor());
Flux.range(1, 10)
.publishOn(customSchedule)
.log("publish thread:")
.flatMap(n -> Mono.fromCallable(() -> n).subscribeOn(Schedulers.parallel()))
.log("subscribe thread:")
.subscribe();
在 Project Reactor 中,Processor
曾是一个关键组件,但随着 Reactor 3.4+ 版本的演进,官方逐渐将其标记为弃用(Deprecated),并推荐使用更现代的 Sink
API 替代。以下是弃用原因、两者核心区别。
Processor
实现(如 DirectProcessor
、UnicastProcessor
)非线程安全,直接调用 onNext
、onComplete
等方法需手动同步。Sink
提供 tryEmitNext
、tryEmitError
等方法,确保多线程推送数据时的安全性。Publisher
和 Subscriber
,这种设计虽然灵活,但导致职责不清晰,容易误用。processor
对背压的支持差异大:
DirectProcessor
完全忽略背压(无界队列)。UnicastProcessor
支持单订阅者的背压,但需手动配置缓冲区。Sink:内置配置,通过 onBackpressureBuffer
、onBackpressureError
等链式方法直接定义背压行为。
onComplete
或 onError
结束流,若遗漏可能导致资源泄漏或订阅者挂起。tryEmitComplete
和 tryEmitError
明确结束流,避免资源泄漏。Processor
的 API 未针对现代响应式编程模式优化(如缺少对重试、重播的内置支持)。Sinks.Many
的 multicast()
、unicast()
或 replay()
快速配置多订阅者行为。 由于processor已经被弃用,不推荐使用,这里不做过多介绍。
1:发送单个数据
Sinks.One sink = Sinks.one();
Mono mono = sink.asMono();
mono.subscribe(
value -> System.out.println("Received: " + value),
error -> System.err.println("Error: " + error),
() -> System.out.println("Completed")
);
sink.tryEmitValue("Hello"); // 等效于 tryEmitNext + tryEmitComplete
2:发送多个数据
// 创建多播 Sink, 并设计缓冲被压策略
Sinks.Many sink = Sinks.many().multicast().onBackpressureBuffer();
//转换为flux
Flux hotFlux = sink.asFlux().map(String::toUpperCase);
// 订阅者A
hotFlux.subscribe(i -> System.out.println("订阅者A: " + i));
// 订阅者B
hotFlux.subscribe(i -> System.out.println("订阅者B: " + i));
// 发送数据
sink.tryEmitNext("hello");
sink.tryEmitNext("world");
sink.tryEmitComplete();
3:支持历史数据
// 创建重播 Sink,保留最近 2 个元素
Sinks.Many sink = Sinks.many().replay().limit(2);
sink.tryEmitNext("A");
sink.tryEmitNext("B");
// 订阅者1 (接收历史数据 A, B)
sink.asFlux().subscribe(s -> System.out.println("Sub1: " + s));
// 推送新数据
sink.tryEmitNext("C");
// 订阅者2(接收历史数据 B, C)
sink.asFlux().subscribe(s -> System.out.println("Sub2: " + s));
//输出
Sub1: A
Sub1: B
Sub1: C
Sub2: B
Sub2: C
onBackpressureBuffer
(缓冲策略)ERROR
:缓冲区满时抛出 IllegalStateException
。DROP_LATEST
:丢弃新数据,保留旧数据。DROP_OLDEST
:丢弃最旧数据,保留新数据。onBackpressureError
(错误策略)IllegalStateException
)。directBestEffort
(尽力而为策略)replay
(重播策略)行为
新订阅者重播历史数据
同时支持实时数据推送。
可配置重播的缓冲区大小(如保留最近的 N
个元素)。
背压处理
onBackpressureBuffer
或 directBestEffort
策略。适用场景:需要新订阅者获取历史数据的场景。
multicast()
:默认使用 directBestEffort
(无缓冲区)。unicast()
:默认使用 onBackpressureBuffer
(无界缓冲区)。replay()
:默认保留所有历史数据(无界缓冲区)。在 Project Reactor 中,Hooks 是一组全局回调机制,允许对 Reactor 库的默认行为进行定制化扩展,用于调试、监控或修改响应式流的执行逻辑。
1. onOperatorError
作用:捕获操作符执行过程中抛出的未处理异常。
典型场景:统一日志记录、转换错误类型。
Hooks.onOperatorError((error, context) -> {
System.err.println("全局捕获异常: " + error);
return error;
});
2. onNextDropped
作用:处理因下游取消订阅、背压溢出等原因被丢弃的 onNext
元素。
典型场景:记录丢失的数据,用于审计或补偿。
Hooks.onNextDropped(item ->
System.out.println("元素被丢弃: " + item)
);
3. onErrorDropped
作用:处理因下游已终止(如已调用 onComplete
)而被丢弃的 onError
信号。
典型场景:避免静默忽略错误。
Hooks.onErrorDropped(error ->
System.err.println("错误被丢弃: " + error)
);
4. onOperatorDebug
作用:启用调试模式,为异步操作符生成增强的堆栈跟踪信息(含订阅点位置)。
代价:增加性能开销,仅限开发环境使用。
Hooks.onOperatorDebug(); // 启用调试模式
5. onEachOperator
/ onLastOperator
作用:在每个操作符执行前后插入自定义逻辑(如日志、指标采集)。
典型场景:性能监控、动态修改数据流。
Hooks.onEachOperator(operator -> {
long start = System.currentTimeMillis();
return original -> original.doFinally(signal ->
System.out.println("操作符耗时: " + (System.currentTimeMillis() - start) + "ms")
);
});
6. 重置 Hooks
恢复默认行为
javaCopy CodeHooks.resetOnOperatorError();
Hooks.resetOnNextDropped();
Hooks.resetOnOperatorDebug();
在 Project Reactor 中,Context
是用于在响应式流的各个阶段之间传递上下文数据的核心机制。它解决了传统 ThreadLocal
在异步、多线程环境中的局限性,允许数据在操作符链中安全传递。以下是 Context
的详细解析,涵盖其设计思想、API 使用及典型场景。
ThreadLocal
无法跨线程传递。Context
提供一种与订阅链绑定的、不可变的键值存储,确保上下文数据在流的生命周期内可被安全访问。Map
结构,支持类型安全的键(ContextKey
)。contextWrite
会覆盖先调用的。Context
优先被访问。通过 contextWrite
操作符将 Context
写入响应式流,通过deferContextual
在流中读取 Context
//注意由于ontext的传递是从底部往上传递的,所以必须在下面(A点)先写入才能在(B点读取到)
Flux.just("A", "B", "C", "D")
//记为B点 拼接 Context 中的值
.flatMap(s -> {
System.out.println("ssss:" + s);
return Mono.deferContextual(ctx -> Mono.just(s + ctx.get("suffix")));
}
)
//记为A点 写入 Context(关键:必须在读取操作之前调用)
.contextWrite(Context.of("suffix", "-ctx"))
// 订阅输出结果
.subscribe(System.out::println);
Context自底向上(Downstream → Upstream)传播示例
由于Context自底向上的传播特性,所以Context中B点的值会覆盖A点的值
Flux.just("A", "B", "C", "D")
// 拼接 Context 中的值
.flatMap(s -> {
//由于ctx222会覆盖ctx111,所以此处拼接的是ctx222
System.out.println("ssss:" + s);
return Mono.deferContextual(ctx -> Mono.just(s + ctx.get("suffix")));
}
)
//记为B点, 写入 Context ctx222会覆盖ctx111
.contextWrite(Context.of("suffix", "-ctx222"))
//记为A点, 写入 Context
.contextWrite(Context.of("suffix", "-ctx111"))
// 订阅输出结果
.subscribe(System.out::println);
//输出
ssss:A
A-ctx222
ssss:B
B-ctx222
ssss:C
C-ctx222
ssss:D
D-ctx222
通过深入理解Project Reactor这些核心概念,可以更好地驾驭响应式编程范式,构建出更高效、更弹性的分布式系统。与现代虚拟线程的结合,为构建新一代高并发应用提供了更优解。通过合理选择调度策略、优化线程模型,可以在保持代码简洁的同时,充分发挥硬件性能。
路漫漫其修远兮,吾将上下而求索
参与评论
手机查看
返回顶部