当前位置：首页 > news >正文

响应式编程-Project Reactor Mono 介绍

news 2025/7/8 0:36:26

响应式编程-Project Reactor Mono 介绍

本文以Mono的角度来介绍Reactor编程，Flux的使用同理。

初体验

Web应用 controller 方法在Spring webmvc 和 Spring webFlux下Controller方法实现示例如下：

Spring webmvc:

@GetMapping("/test1")

@ResponseBody

public String test1(){

String result = geterateTest();

return result;

}

Spring webFlux

@GetMapping("/test2")

@ResponseBody

public Mono<String> test2(){

Mono<String> result = Mono.fromSupplier(this:: geterateTest);

return result;

}

一个的响应是String对象，另一个是Mono<String>对象。Reactor Mono表示一个产生0-1元素的异步序列，异步指Mono创建的时候并不会执行任何操作，当Mono发生订阅时才触发Mono序列的运行。非阻塞表示test2方法不会产生任何阻塞，即使genereateTest里面是一个阻塞的操作，因为此时不会执行实际的逻辑，所以不会发生任何阻塞。

NettyHttpServer.onStateChange方法中构建Mono并进行订阅。

HttpServerOperations ops = (HttpServerOperations)connection;

//Web Flux将按照Spring Web中的约定构建一个Publisher（执行过滤器、Controller方//法）

Publisher<Void> publisher = (Publisher)this.handler.apply(ops, ops);

Mono<Void> mono = Mono.deferContextual((ctx) -> {

ops.currentContext = Context.of(ctx);

return Mono.fromDirect(publisher);

});

……

//subscribe将触发前面Spring web中封装在Mono构建过程中的业务逻辑的真正执行。

//如果我们按照命令是编程去编写代码，业务逻辑在构建Mono的过程中就执行了。

mono.subscribe(ops.disposeSubscriber());

注： Spring web flux框架下也可以按照传统的命令式编程。

Mono的构建

Reactor编程可以分为异步序列Mono/Flux的构建和和使用两部分。

Mono的基本构建

Mono类提供了大量静态方法帮助构建Mono。

just(T):返回T类型对象的Mono序列
fromFuture(future):Mono序列的元素对象由future产生，订阅时Future产生T并推送至订阅者。其他from方法类似。
empty():返回一个订阅时直接完成的异步序列
error():返回一个订阅时直接推送错误信号的序列

其他方法详见Mono类API：

如：Mono<String> mono = Mono.just("TEST");

Mono装配

假设我们按照上面示例，将整个程序都以响应式编程的模式进行开发，方法都返回一个异步序列Mono/Flux。当调用者调用某一个方法时，面对返回的Mono/Flux对象有两种选择：1. 订阅（触发执行）， 2.装配（Assembly）：继续将获取到的异步序列封装到一个新的异步序列中，继续返回给外部调用者。如：Spring Web Flux 则是将Spring web 定义的包括WebFilter、Controller等逻辑组装成一个复合的Mono，最终进行订阅。

图1 Mono装配示例

OptimizableOperator 接口

OptimizableOperator <IN, OUT>接口提供了指向下一个OptimizableOperator的指针，并且提供了从IN型订阅者获取OUT订阅者的方法，提供了一个Mono串行的组装方法。

图2 OptimizableOperator接口串行组装示意图

要实现一个串行化的Mono组装类通常实现抽象类InternalMonoOperator<I, O>，构造函数传入一个Mono<I>,得到一个新的O型序列。实现subscribeOrReturn方法将O型订阅转化为原I型订阅者，新的I型订阅者实现了基于O性订阅者之上的强化操作。Mono提供了大量InternalMonoOperator<I,O>的实现类。下面对MonoFilter进行分析，解释了如果创建基于InternalMonoOperator实现的装配类和使用方法。

MonoFilter

将原Mono上增加一个过滤Predicate函数，当原Mono产生元素时，只有Predicate测试通过的元素才会传递给最终的订阅者，测试失败将进行过滤，Mono元素直接完成。

final class MonoFilter<T> extends InternalMonoOperator<T, T> {

final Predicate<? super T> predicate;

//构造函数必须包含源Mono，和其他附加增加元素，这里是一个Predicate函数

MonoFilter(Mono<? extends T> source, Predicate<? super T> predicate) {

super(source);

this.predicate = Objects.requireNonNull(predicate, "predicate");

}

/**

* 实现subscribeOrReturn，接收新Mono类型的订阅者，返回原Mono类型的订阅者。

* 新的订阅者实现订阅时装配的目的，这里只有通过Predicate函数测试的元素，才会

* 调用actual.onNext(T)方法推送给最终的订阅者

**/

@Override

@SuppressWarnings("unchecked")

public CoreSubscriber<? super T> subscribeOrReturn(CoreSubscriber<? super T> actual) {

if (actual instanceof ConditionalSubscriber) {

return new FluxFilter.FilterConditionalSubscriber<>((ConditionalSubscriber<? super T>) actual, predicate);

}

return new FluxFilter.FilterSubscriber<>(actual, predicate);

}

......

}

Mono内置了大量的InternalMonoOperator实现类，如MonoFilter，但Reactor框架并不对外暴露这些类，（这些实现类都是包内可见的），而是通过Mono方法的形式去方便获取各个可实现类的对象，并且统一以Mono类型的对外暴露。抽象统一的Mono使用范式比起暴露各种各样的实现细节显得简洁清晰。

我们可以使用Mono内置的InternalMonoOperator实现类，也可以实现自己的InternalMonoOperator类，但应和Reactor框架保持统一的用法，在Mono的使用上统一以Mono类型和协议进行操作，不对外暴露具体的实现细节。

Mono 提供的装配方法

Reactor框架并不暴露具体的装配类细节，而是提供了大量静态或实例方法来对Mono进行装配，返回装配后的新Mono。如上节所述的MonoFilter使用方法如下：

Mono.just(2).filter( (v -> v % 2 != 0)).subscribe(i -> System.out.println(i),

error -> System.err.println("Error: " + error),

()-> System.out.println("complete"));

Mono filter方法返回了一个可以对原序列元素进行检测的增强Mono，上述例子因Mono.just(2) 中的元素值2 无法通过（v -> v % 2 != 0）的测试，将被过滤掉，无法传给最终的订阅者，而只能接受到原序列的结束信号，因此只会打印“complete“。

Filter方法显示实际是返回的MonoFilter对象。

public final Mono<T> filter(final Predicate<? super T> tester) {

……

return onAssembly(new MonoFilter<>(this, tester));

}

其他Mono装配方法：

Mono<Tuple2<T1, T2>> zip(Mono<? extends T1> p1, Mono<? extends T2> p2)

:将一个T1类型元素的Mono和一个T2类型元素的Mono中的元素组合成一个Tuple2<T1,T2>元素的Mono. Mono还提供了zip的多种版本，满足各种情况的Mono组合模式。

public final Mono<T> timeout(Duration timeout)： 当原序列产生一个T类型元素后，如果没有在指定的时间内完成，则将触发一个错误。如果在限期内完成则没有任何影响，该实现使用了MonoTimeout<T, U, V> extends InternalMonoOperator<T, T>。
doOnXXXX系列方法，如doOnCancel, doOnNext, doOnError等，返回在特定事件上加入行为的增强Mono。

更多Mono的装配方法详见Mono API。

Mono的使用

Mono的使用其实只有一种就是对Mono进行订阅，但是Mono类也提供了其他传统的接口来进行Mono的使用。

Mono的订阅

订阅Mono很简单，调用Mono对象的subscribe方法，传入一个CoreSubscriber的实现对象即可。

Mono.subscribe.源码中展示了对Mono装配后的复合Mono进行订阅的处理逻辑。

public final void subscribe(Subscriber<? super T> actual) {

//获取最后一个装配的Mono corePublisher

CorePublisher publisher = Operators.onLastAssembly(this);

CoreSubscriber subscriber = Operators.toCoreSubscriber(actual);

......

//如果最后一个装配的publisher 实现了OptmizableOperator接口，一路组装

//增强的Subscriber，按照循序后去下一个OptmizableOperator

if (publisher instanceof OptimizableOperator) {

OptimizableOperator operator = (OptimizableOperator) publisher;

while (true) {

subscriber = operator.subscribeOrReturn(subscriber);

if (subscriber == null) {

return;

}

OptimizableOperator newSource = operator.nextOptimizableSource();

if (newSource == null) {

publisher = operator.source();

break;

}

operator = newSource;

}

//直到最底层的CorePublisher,使用最终转换所得的subscriber进行订阅，

//原始序列产生的序号，将在一些列增强subscriber的增强下，或丢弃、或加工后传给

//实际的订阅者

publisher.subscribe(subscriber);

}

Mono的简化使用

Mono 提供了一些方法简化Mono的订阅操作，如block() 阻塞当前线程知道Mono序列返回元素或完成/异常信号

PublishOn和SubscribeOn

publishOn 和 SubscribeOn 传入Scheduler对象，将Mono的行为交由Scheduler的现成执行。其中publishOn调用之后的序列行为在新的执行线程执行，而SubscribeOn则是整个序列的执行都在新的现成中执行。

final Flux<String> flux = Flux

.range(1, 2)

.map(i -> 10 + i)

.publishOn(s)

.map(i -> "value " + i);

flux.subscribe(System.out::println)

final Flux<String> flux = Flux

.range(1, 2)

.map(i -> 10 + i)

.subscribeOn(s)

.map(i -> "value " + i);

flux.subscribe(System.out::println)

总结

本文对Reactor的Mono编程进行了初步的介绍，体现了响应式编程的核心在于异步序列的构建（Mono/Flux）和订阅使用。其中构建时对Mono/Flux的装配（Assembly）是整个编程模型的核心。