• 响应式编程-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 test2(){

            Mono result = Mono.fromSupplier(this:: geterateTest);

            return result;

        }

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

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

    HttpServerOperations ops = (HttpServerOperations)connection;

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

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

    Mono 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 mono = Mono.just("TEST");

    Mono装配

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

    图1 Mono装配示例

    OptimizableOperator 接口

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

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

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

    MonoFilter

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

    final class MonoFilter extends InternalMonoOperator {

             final Predicate predicate;

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

             MonoFilter(Mono source, Predicate predicate) {

                      super(source);

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

             }

             /**

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

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

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

             **/

             @Override

             @SuppressWarnings("unchecked")

             public CoreSubscriber subscribeOrReturn(CoreSubscriber actual) {

                      if (actual instanceof ConditionalSubscriber) {

                              return new FluxFilter.FilterConditionalSubscriber<>((ConditionalSubscriber) 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 filter(final Predicate tester) {

             ……

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

    }

    其他Mono装配方法：

    • Mono> zip(Mono p1, Mono p2)

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

    • public final Mono timeout(Duration timeout)： 当原序列产生一个T类型元素后，如果没有在指定的时间内完成，则将触发一个错误。如果在限期内完成则没有任何影响，该实现使用了MonoTimeout extends InternalMonoOperator。
    • 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 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 flux = Flux

        .range(1, 2)

        .map(i -> 10 + i) 

        .publishOn(s) 

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

    flux.subscribe(System.out::println)

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

  • 相关阅读:
    研发过程中的文档管理与工具
    【微服务容器化】第五章-Dockerfile
    竞赛 基于大数据的时间序列股价预测分析与可视化 - lstm
    ARM hint instruction-WFI(Wait For Interrupt)指令详解
    C++:类与面向对象&static和this关键字&其他关键字
    金仓数据库KingbaseES数据库参考手册(服务器配置参数4. 连接和认证)
    自定义Graph Component：1.2-其它Tokenizer具体实现
    C++深度优先搜索(DFS)算法的应用：树中可以形成回文的路径数
    Windbg 命令 （一）
    3 Minute Thesis (3MT)
  • 原文地址：https://blog.csdn.net/dreamsofa/article/details/134274084