步入响应式编程篇（二）之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)，以此类推。
默认情况下，发布者使用的线程就是订阅者的线程，那就证明一下：
如上用log()打印出，订阅者和发布者使用的线程都是main，那如果订阅者开启新的线程，下图也能看到发布者回调onNext操作也是使用订阅者的线程：
下图，使用publishOn()里指定新线程Schedulers.single()，代表过滤后，发布者使用新线程会回调onNext()
图中①是filter之前，发布者还是用的订阅者的线程回调onNext，②是filter之后，发布者使用新的线程回调onNext()。
总之，上面只是介绍reactor api操作的冰山一角，至于更多操作可以查看官网哈
如有需要收藏的看官，顺便也用发财的小手点点赞哈，如有错漏，也欢迎各位在评论区评论！
参考官网：https://projectreactor.io/docs/core/release/reference/aboutDoc.html
![[JavaScript] 面向对象编程](https://i-blog.csdnimg.cn/img_convert/17730a1dadec799781c9255c287f35ab.gif)
