响应式编程-Project Reactor Mono 介绍

响应式编程-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）是整个编程模型的核心。