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

步入响应式编程篇（二）之Reactor API

article 2025/1/24 6:12:43

步入响应式编程篇（二）之Reactor API

前言
回顾响应式编程
Reactor API的使用
- Stream
- 引入依赖
- Reactor API的使用
- - 流源头的创建
- reactor api的背压模式
- 发布者与订阅者使用的线程
- - 查看弹珠图
  - 查看形成新流的日志

前言

对于响应式编程的基于概念，以及JDK自带的落地实现，可以查看步入响应式编程篇（一）
本篇将介绍Reactor API的使用：

reactor官网介绍，响应式编程是一种与数据流和变化传播相关的异步编程范式。这意味着可以通过所采用的编程语言轻松表达静态（例如数组）或动态（例如时间发射器）数据流；

对比Flow api以及completableFuture，前者编写代码比较麻烦，编写处理器还要自定义一个类，后者还不能满足响应式编程，两者都有其局限性，而Reactor API是基于Stream流操作的，无论是编写还是响应式编程都能满足；

回顾响应式编程

①在面向对象语言中，反应式编程范式通常作为观察者设计模式的扩展。还可以比较主要的反应流模式与熟悉的迭代器设计模式，因为 Iterable- 所有这些库中的迭代器对。一个主要的区别是，虽然Iterator是基于拉的，但反应流是基于推的。

②使用迭代器是一种命令式编程模式，即使访问值的方法完全由Iterable负责。实际上，由开发人员来选择何时访问序列中的下一个（）项。在反应式流中，上述对的等价物是发布者-订阅者。但它是发布者在新的可用值到来时通知订阅者，这种推送方面是响应的关键。此外，应用于推送值的操作是以声明方式而不是命令方式表达的：程序员表达计算的逻辑，而不是描述其确切的控制流。

③除了推送值之外，还以定义良好的方式涵盖了错误处理和完成方面。发布者可以将新值推送到其订阅者（通过调用onNext），但也可以发出错误信号（通过调用onError）或完成信号（通过调用onComplete）。错误和完成都会终止序列。这可以归纳如下：

onNext x 0…N [onError | onComplete]

这种方法非常灵活。该模式支持没有值、只有一个值或有n个值的用例（包括无限序列的值，例如时钟的连续滴答声）。

Reactor API的使用

Stream

用的就是Java 8引用的Stream API，使用Stream api时也会结合lambda表达式和函数式接口，所以Stream api是包括它们的；
虽然是众所周知，但笔者在此还是提一嘴，Stream API的链式调用，是每个元素完成整个流所有步骤的处理后，再遍历下一个元素的。而不是所有元素完成后，再让流中的下一个步骤处理全部元素，这是一不小心就会陷入的理解误区；

还有诸如flatMap()、map()等参数为函数式接口的种类，总共有四类
//Predicate 有入参且返回值固定为boolean
//Consumer 有入参无返回值
//Function 有一个入参且一个返回值
//Supplier 无入参且有返回值

引入依赖

需引入Reactor API的依赖

<dependency>
    <groupId>io.projectreactor</groupId>
    <artifactId>reactor-bom</artifactId>
    <version>2023.0.0</version>
</dependency>

Reactor API的使用

Reactor API的使用是定义流源头，然后往下流，使用链式处理多个步骤，最后流向最后，就是异常或响应信号。

流源头的创建

创建流的方式有Flux和Mono两种，前者是用于创建多个元素的流，后者是创建只有一个元素的流；
在这里插入图片描述

这就分别创建了Flux流和Mono流，很轻易实现了一个全异步的系统。相比Flow流的一顿操作，是不是简便了很多；

而创建流，需指定流的源头，有四种方式
①just()，如上例，穷举出所有元素；
②rang()；

//表示流有1到100个元素
Flux.range(1,100);

③generate();
这适用于同步和逐个发射，这意味着接收器是一个SynchronousSink，并且它的next（）方法在每次回调调用中最多只能被调用一次。

//入参是泛型为SynchronousSink的Consumer接口
Flux.generate(sink->{
   //查库存
   sink.next(queryStock("1"));
   sink.complete();
}).subscribe(System.out::println);


//但同一次回调中循环调用next()会报错，所以先初始化state值为0
Flux.generate(()->0,(state,sink)->{
            if (state<=10) {
            	//小于10，则继续调用，里面的逻辑是每次将state+1
                sink.next(state);
            }else{
                sink.complete();
            }
            return state+1;
        }).subscribe(System.out::println);

④create方式，create是一种更高级的Flux程序化创建形式，它适合于每轮多个发射，甚至来自多个线程。create无需指定初始值而且可以多次执行sink#next()：

Flux.create(sink->{
    for (int i=0;i<2;i++) {
        sink.next(queryStock("1"));
    }
    sink.complete();
}).subscribe(System.out::println);

在业务上通常使用后两种，例如可以从缓存或DB查到数据，然后类似于观察者模式，往后推送处理；更多使用create，因为generate需设置初始值，而且每次回调只能调用一次sink#next()

reactor api的背压模式

使用背压模式的原因——响应式编程的思想，主要在于推，推送到每个处理器操作，如果不管当前处理能力就很容易处理出问题，类似于MQ消费者控制从队列中获取数据量，要控制消费能力，于是引出背压模式，接收时通过request告诉上游，所以在Reactor中实现背压时，消费者压力传播回源的方式是向上游操作员发送请求。当前请求的总和有时被称为当前“需求”或“待决请求”。需求的上限为Long.MAX_VALUE，表示一个无限的请求（意思是“尽可能快地生产”-基本上禁用反压）。
如下，与Flow API类似，也是在自定义订阅者中指定背压请求request()

Flux.range(1, 10)
    .doOnRequest(r -> System.out.println("request of " + r))
    .subscribe(new BaseSubscriber<Integer>() {

      @Override
      public void hookOnSubscribe(Subscription subscription) {
        request(1);
      }

      @Override
      public void hookOnNext(Integer integer) {
        System.out.println("Cancelling after having received " + integer);
        cancel();
      }
    });

发布者与订阅者使用的线程

查看弹珠图

Flux.create(sink->{
    for (int i=0;i<2;i++) {
        sink.next(i*queryStock("1"));
    }
    //代表推送给下一个操作形成新流
    sink.complete();
    //过滤不等于0的数
}).filter(Predicate.not(Predicate.isEqual(0))).subscribe(System.out::println);

鼠标点到filter操作，就可以看到元素（这些不同形状的方块就是了）经过filter后，变少了，点到每个操作map、flatmap等都可以看到对应的弹珠图，可根据弹珠图快速理解形成新流的操作
在这里插入图片描述

查看形成新流的日志

在这里插入图片描述
在每个流后面使用log()，查看该流形成每个元素的日志，如都是request无限元素，然后create后，就调onNext(0)，形成元素0，后面filter就过滤了0，就没有再调onNext(0)，继续create元素1时，就调 onNext(1)，形成元素1，后面filter后，形成的新流中仍然有元素1，所以还会调用onNext(1)，以此类推。