步入响应式编程篇（二）之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+1sink.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>() {@Overridepublic void hookOnSubscribe(Subscription subscription) {request(1);}@Overridepublic 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)，以此类推。

默认情况下，发布者使用的线程就是订阅者的线程，那就证明一下：
如上用log()打印出，订阅者和发布者使用的线程都是main，那如果订阅者开启新的线程，下图也能看到发布者回调onNext操作也是使用订阅者的线程：

下图，使用publishOn()里指定新线程Schedulers.single()，代表过滤后，发布者使用新线程会回调onNext()

图中①是filter之前，发布者还是用的订阅者的线程回调onNext，②是filter之后，发布者使用新的线程回调onNext()。

总之，上面只是介绍reactor api操作的冰山一角，至于更多操作可以查看官网哈

如有需要收藏的看官，顺便也用发财的小手点点赞哈，如有错漏，也欢迎各位在评论区评论！

参考官网：https://projectreactor.io/docs/core/release/reference/aboutDoc.html