Reactor核心API-工具盒子

响应式编程总结：

底层：基于数据缓冲队列 + 消息驱动模型 + 异步回调机制（事件驱动）
编码：流式编程 + 链式调用 + 声明式API
效果：优雅全异步 + 消息实时处理 + 高吞吐量 + 占用少量资源

响应式编程准备

在reactor文档中看一下如何使用Reactor框架。

第一步，先导入依赖

<dependencyManagement> 
   <dependencies>
       <dependency>
           <groupId>io.projectreactor</groupId>
           <artifactId>reactor-bom</artifactId>
           <version>2023.0.6</version>
           <type>pom</type>
           <scope>import</scope>
       </dependency>
   </dependencies>
</dependencyManagement>

\<dependencies\>
\<dependency\>
\<groupId\>io.projectreactor\</groupId\>
\<artifactId\>reactor-core\</artifactId\>

`</dependency>
<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
`

事件感知`API`

链式API中，下面的操作符，操作的是上面的流

当流发生什么事的时候，触发一个回调，系统调用提前定义好的钩子函数(Hook,钩子函数)。doOnXxx

doOnComplete，在流成功完成时触发

Flux<Integer> flux = Flux.just(1, 2, 3).doOnComplete(()-> System.out.println("流已结束"));
flux.subscribe(System.out::println);

输出结果：

doOnCancel，流被取消时触发

.doOnCancel(()-> System.out.println("流已取消"))

doOnError，流出错时触发

.doOnError(throwable -> System.out.println("流出错"))

doOnNext，流的每个元素（流的数据）到达时触发

.doOnNext(integer -> System.out.println("处理数据" + integer))

doOnEach，流的每个元素（流的数据和信号）到达的时候触发
```
.doOnEach(integerSignal -> System.out.println("处理数据" + integerSignal))
```
针对流中的每个信号（元素、错误信号或完成信号）的操作。这意味着它在每个元素、错误或完成信号发出时都会执行指定的操作。

.doOnRequest，消费者请求流元素的时候触发

.doOnRequest(integer -> System.out.println("请求数据" + integer))

doOnSubscribe，流被订阅的时候触发

.doOnSubscribe(subscription -> System.out.println("订阅流" + subscription))

doOnDiscard，流被忽略的时候触发

.doOnDiscard(Object.class, integer -> System.out.println("丢弃数据" + integer))

响应式流日志

.log()方法可以把流的过程打印出来，但是位置不同，记录的流也不同。

比如：

Flux.range(1,5)             // 初始流
        .log()
        .filter(i -> i>3)   // 新流1
        .map(i -> "前缀" + i) // 新流2
        .subscribe(System.out::println);

这段代码的执行顺序是：

有一个元素为1~5的初始流，打印这个初始流日志
过滤初始流中大于3的元素，组成新流1
把新流1中的元素加上"前缀"，形成新流2
订阅新流2并输出新流2中的元素

执行结果：

所以这个log()日志记录的是原始流的完整流程。从订阅onSubscribe到请求元素request再到元素到达onNext最后订阅结束onComplete。其中onNext方法执行了5次。证明记录的是初始流。

如果把log()的位置换到下面：

Flux.range(1,5)
        .filter(i -> i>3)
        .log()
        .map(i -> "前缀" + i)
        .subscribe(System.out::println);

执行结果：

这里的日志onNext方法记录的就是4和5，也就是新流1的完整流程。

请求重塑`buffer`缓冲

Flux.range(1,5)
        .buffer(3)
        .log()
        .subscribe(System.out::println);

有一个元素为1~5的流，缓冲区空间为3个元素，装满3个元素后再一起发给消费者，如果到最后只剩2个元素了，不满3个元素，那就把剩下的2个元素一起发给消费者。通过日志可以看到：onNext方法只被调用了两次，说明只收到了两次元素。

这时候，如果消费者这时候request(1)请求一次数据，那么收到的将会是[1, 2, 3]，这一次数据是3个元素。

请求重塑`limitRate`限流

Flux.range(1,5)
        .log()
        .limitRate(2)
        .subscribe();

限流操作，每次向上游请求多少个元素。但是这里有一个阈值，是一个75%的预取策略。如果75%的元素已经处理了，继续请求新的75%的元素。

比如.limitRate(100)第一次请求100个数据，如果75个元素已经处理了，继续请求新的75个元素。

创建序列`generate`-同步环境

以编程方式创建自定义序列。编程时有可能有一些序列不是通过Flux、Mono这些API简单创建出来的，而是需要经过一系列的代码操作。这个时候，同步环境下可以用generate。Flux和Mono都有这个方法。

Sink是流通道，如果往流通道里塞一个元素，也代表着流里面有一个元素了。调用.next方法，就代表往通道里放了一个元素。

Flux.generate(() -> 0,  // 初始state值
        (state, sink) -> {
            if (state <= 10) {
                if (state == 7){
                    sink.next(state + 70);
                }else {
                    sink.next("元素： " + state);  // 把元素传出去
                }
            }else {
                sink.complete();    // 完成信号
            }
            return state + 1;
        })
        .log()
        .subscribe();

执行结果：

这就是自定义的序列，在元素为7时换成77。

创建序列`create`-多线程

同步环境下可以用generate，异步多线程的情况下就用create。

Flux.create(fluxSink -> {
    for (int i = 0; i <= 10; i++ ){
        fluxSink.next(i);
    }
        })
        .log()
        .subscribe();

执行结果：

自定义元素处理`handle`

handle可以自定义处理流中元素处理规则。

Flux.range(1,5)
        .handle((value, sink)->{
            System.out.println("拿到的值：" + value);
            sink.next("自定义处理结果：P" + value);
        })
        .log()
        .subscribe();

执行结果：

与map相似，但是map必须是映射成同类型的元素，而handle可以任意操作。handle后，流中可能有不同类型的数据。

自定义线程调度

默认是用当前线程，生成整个流、发布流、流操作

publishOn：改变发布者所在线程池

subscribeOn：改变订阅者所在线程池

Schedulers.immediate()，当没有指定的时候默认用的就是这个方法：无执行上下文。既当前线程运行所有操作
```
Flux.range(1,5)
       .publishOn(Schedulers.immediate())
       .log()
       .subscribe();
```
执行结果：都是同一个线程

Schedulers.single()，使用固定的一个单线程执行所有操作

Flux.range(1,5)
       .publishOn(Schedulers.single())
       .log()
       .subscribe();

执行结果：

Schedulers.boundedElastic()，使用固定大小的线程池执行所有操作。默认是：线程池中有 10*cpu核心个数的线程，队列默认为100k，KeepAliveTime默认为60s。
```
Flux.range(1,5)
       .publishOn(Schedulers.boundedElastic())
       .log()
       .subscribe();
```
执行结果：

Schedulers.parallel()，并发池

Flux.range(1,5)
       .publishOn(Schedulers.parallel())
       .log()
       .subscribe();

执行结果：

Schedulers.fromExecutor(new ThreadPoolExecutor(4,8,60, TimeUnit.SECONDS,new LinkedBlockingQueue<>(1000)))，自定义线程池。fromExecutor方法里的参数是自己自定义的。
```
Flux.range(1,5)
       .publishOn(Schedulers.fromExecutor(new ThreadPoolExecutor(4,8,60, TimeUnit.SECONDS,new LinkedBlockingQueue<>(1000))))
       .log()
       .subscribe();
```
执行结果：