响应式编程 Reactor 学习小记

时间 2019-11-06

原文原文链接

从响应式编程提及

响应式编程是一种关注于数据流（data streams）和变化传递（propagation of change）的异步编程方式。这意味着它能够用既有的编程语言表达静态（如数组）或动态（如事件源）的数据流。html

在响应式编程方面，微软跨出了第一步，它在 .NET 生态中建立了响应式扩展库（Reactive Extensions library, Rx）。接着 RxJava 在 JVM 上实现了响应式编程。后来，在 JVM 平台出现了一套标准的响应式编程规范，它定义了一系列标准接口和交互规范。并整合到 Java 9 中（Flow 类）。java

响应式编程一般做为面向对象编程中的“观察者模式”（Observer design pattern）的一种扩展。响应式流（reactive streams）与“迭代子模式”（Iterator design pattern）也有相通之处，由于其中也有 Iterable-Iterator 这样的对应关系。主要的区别在于，Iterator 是基于 “拉取”（pull）方式的，而响应式流是基于“推送”（push）方式的。react

iterator 是一种“命令式”（imperative）编程范式，即便访问元素的方法是 Iterable 的惟一职责。关键在于，何时执行 next() 获取元素取决于开发者。
响应式流中，相对应的角色是 Publisher-Subscriber，可是当有新的值到来的时候，却反过来由发布者（Publisher）通知订阅者（Subscriber），这种“推送”模式是响应式的关键

此外，对推送来的数据的操做是经过一种声明式（declaratively）而不是命令式（imperatively）的方式表达的：开发者经过描述“控制流程”来定义对数据流的处理逻辑。git

除了数据推送，对错误处理（error handling）和完成（completion）信号的定义也很完善。一个 Publisher 能够推送新的值到它的 Subscriber（调用 onNext 方法），一样也能够推送错误（调用 onError 方法）和完成（调用 onComplete 方法）信号。错误和完成信号均可以终止响应式流。能够用下边的表达式描述：github

onNext x 0..N [onError | onComplete]
复制代码

这种方式很是灵活，不管是有/没有值，仍是 n 个值（包括有无限个值的流，好比时钟的持续读秒），均可处理。编程

以上来自 projectreactor.io/docs/core/r… 翻译数组

Reactive Streams

Reactive Streams 是上面提到的一套标准的响应式编程规范。它由四个核心概念构成：bash

消息发布者：只有一个 subscribe 接口，是订阅者调用的，用来订阅发布者的消息。发布者在订阅者调用 request 以后把消息 push 给订阅者。
```
public interface Publisher<T> {
    public void subscribe(Subscriber<? super T> s);
}
复制代码
```
订阅者：订阅者包括四个接口，这些接口都由 Publisher 触发调用的。onSubscribe 告诉订阅者订阅成功，并返回了一个 Subscription ；经过 Subscription 订阅者能够告诉发布者发送指定数量的消息（request 完成）；onNext 是发布者有消息时，调用订阅者这个接口来达到发布消息的目的；onError 通知订阅者，发布者出现了错误；onComplete 通知订阅者消息发送完毕。
```
public interface Subscriber<T> {
    public void onSubscribe(Subscription s);
    public void onNext(T t);
    public void onError(Throwable t);
    public void onComplete();
}
复制代码
```

订阅：包括两个接口，请求 n 个消息和取消这次订阅。

public interface Subscription {
    // request(n)用来发起请求数据,其中n表示请求数据的数量,它必须大于0,
    // 不然会抛出IllegalArgumentException,并触发onError,request的调用会
    // 累加,若是没有终止,最后会触发相应次数的onNext方法.
    public void request(long n);
    // cancel至关于取消订阅,调用以后,后续不会再收到订阅,onError 和 
    // onComplete也不会被触发
    public void cancel();
}
复制代码

处理器：Processor 同时继承了 Subscriber 和 Publisher；其表明一个处理阶段。
```
public interface Processor<T, R> extends Subscriber<T>, Publisher<R> {
}
复制代码
```

Reactive Streams 经过上面的四个核心概念和相关的函数，对响应式流进行了一个框架性的约定，它没有具体实现。简单来讲，它只提供通用的、合适的解决方案，你们都按照这个规约来实现就行了。app

Java 的 Reactive Programming 类库主要有三个，分别是 Akka-Streams ，RxJava 和 Project Reactor。Spring 5 开始支持 Reactive Programming，其底层使用的是 Project Reactor。本篇主要是对 Project Reactor 中的一些点进行学习总结。框架

Project Reactor

Project Reactor 是一个基于 Java 8 的实现了响应式流规范（Reactive Streams specification）的响应式库。

Reactor 引入了实现 Publisher 的响应式类 Flux 和 Mono，以及丰富的操做方式。一个 Flux 对象表明一个包含 0..N 个元素的响应式序列，而一个 Mono 对象表明一个包含零或者一个（0..1）元素的结果。

Flux 和 Mono

Flux 是生产者，即咱们上面提到的 Publisher，它表明的是一个包含 0-N 个元素的异步序列，Mono能够看作 Flux 的有一个特例，表明 0-1 个元素，若是不须要生产任何元素，只是须要一个完成任务的信号，可使用 Mono。

Flux-包含 0-N 个元素的异步序列

先来看这张图，这里是直接从官方文档上贴过来的。就这张图作下说明，先来关注几个点：

从左到右的时间序列轴
1-6 为 Flux enitted（发射）的元素
上面 6 后面的竖线标识已经成功完成了
下面的 1-3 表示转换的结果
❌ 表示出现了error，对应的是执行了onError
operator : 操做符，声明式的可组装的响应式方法，其组装成的链称为“操做链”

那总体来看就是 Flux 产生元数据，经过一系列 operator 操做获得转换结果，正常成功就是 onCompleted，出现错误就是 onError。看下面的一个小例子：

Flux.just("glmapper","leishu").subscribe(new Subscriber<String>() {
    @Override
    public void onSubscribe(Subscription subscription) {
        // subscription 表示订阅关系
        System.out.println("onSubscribe,"+ subscription.getClass());
        // subscription 经过 request 来触发 onNext
        subscription.request(2);
    }
    @Override
    public void onNext(String s) {
        System.out.println("currrent value is = " + s);
    }
    @Override
    public void onError(Throwable throwable) {
        System.out.println("it's error.");
    }
    @Override
    public void onComplete() {
        System.out.println("it's completed.");
    }
});
复制代码

执行结果:

onSubscribe,class reactor.core.publisher.StrictSubscriber
currrent value is = glmapper
currrent value is = leishu
it's completed. 复制代码

若是在 onSubscribe 方法中咱们不执行 request，则不会有后续任何操做。关于 request 下面看。

Flux 是一个可以发出 0 到 N 个元素的标准的 Publisher，它会被一个 "error" 或 "completion" 信号终止。所以，一个 Flux 的结果多是一个 value、completion 或 error。就像在响应式流规范中规定的那样，这三种类型的信号被翻译为面向下游的 onNext，onComplete和onError方法。

Mono-异步的 0-1 结果

这张图也来自官方文档，和上面 Flux 的区别就是，Mono 最多只能 emitted 一个元素。

Mono.just("glmapper").subscribe(System.out::println);
复制代码

小结

经过上面两段小的代码来看，最直观的感觉是，Flux 至关于一个 List，Mono 至关于 Optional。其实在编程中全部的结果咱们均可以用 List 来表示，可是当只返回一个或者没有结果时，用 Optional 可能会更精确些。

Optional 相关概念可自行搜索 jdk Optional

另外，Mono 和 Flux 都提供了一些工厂方法，用于建立相关的实例，这里简单罗列一下：

// 能够指定序列中包含的所有元素。建立出来的 Flux 
// 序列在发布这些元素以后会自动结束。
Flux.just("glmapper", "leishu");
// 从一个Iterable 对象中建立 Flux 对象,固然还能够是数组、Stream对象等
Flux.fromIterable(Arrays.asList("glmapper","leishu"));
// 建立一个只包含错误消息的序列。
Flux.error(new IllegalStateException());
// 建立一个包含了从 0 开始递增的 Long 对象的序列。其中包含的元素按照指定的间
// 隔来发布。除了间隔时间以外，还能够指定起始元素发布以前的延迟时间。
Flux.interval(Duration.ofMillis(100)).take(10);
// 建立一个不包含任何消息通知的序列。
Flux.never();
// 建立一个不包含任何元素，只发布结束消息的序列。
Flux.empty(); 
// 建立包含从 start 起始的 count 个数量的 Integer 对象的序列
Flux.range(int start, int count);
// Mono 同上
Mono.empty();
Mono.never();
Mono.just("glmapper");
Mono.error(new IllegalStateException());
复制代码

上面的这些静态方法适合于简单的序列生成，当序列的生成须要复杂的逻辑时，则应该使用 generate() 或 create() 方法。

一些概念

Operator：Operator 是一系列函数式的便捷操做，能够链式调用。全部函数调用基本都是 Reactor 的 Operator ，好比 just，map，flatMap，filter 等。
Processor：上面从 Processor 的接口定义能够看出，它既是一个 Subscriber，又是一个 Publisher；Processor 夹在第一个 Publisher 和最后一个 Subscriber 中间，对数据进行处理。有点相似 stream 里的 map，filter 等方法。具体在数据流转中， Processor 以 Subscriber 的身份订阅 Publisher 接受数据，又以 Publisher 的方式接受其它 Subscriber 的订阅，它从本身订阅的 Publisher 收到数据后，作一些处理，而后转发给订阅它的 Subscriber。
back pressure：背压。对 MQ 有了解的应该清楚，消息积压通常是在消费端，也就是说生产端只负责生产，并不会关心消费端的消费能力，这样就到致使 pressure 积压在消费端，这个是正向的。从上面对 Reactor 中的一些了解，Subscriber 是主动向 Publisher 请求的，这样当消费端消费的速度没有生产者快时，这些消息仍是积压在生产端；这种好处就是生产者能够根据实际状况适当的调整生产消息的速度。
Hot VS Cold ：参考 Hot VS Cold

核心调用过程

Reactor 的核心调用过程大体能够分为图中的几个阶段

声明：不管是使用 just 或者其余什么方式建立反应式流，这个过程均可以称之为声明，由于此时这些代码不会被实际的执行。
subscribe：当调用 subscribe 时，整个执行过程便进入 subscribe 阶段，通过一系列的调用以后，subscribe 动做会代理给具体的 Flux 来实现。
onSubscribe：onSubscribe 阶段指的是 Subscriber#onSubscribe 方法被依次调用的阶段。这个阶段会让各 Subscriber 知道 subscribe 方法已被触发，真正的处理流程立刻就要开始。
request：onSubscribe 阶段是表示订阅动做的方式，让各 Subscriber 知悉，准备开始处理数据。当最终的 Subscriber 作好处理数据的准备以后，它便会调用 Subscription 的 request 方法请求数据。
onNext：经过调用 Subscriber 的 onNext 方法，进行真正的响应式的数据处理。
onComplete：成功的终端状态，没有进一步的事件将被发送。
onError：错误的终端状态（和 onComplete 同样，当发生时，后面的将不会在继续执行）。

消息处理

当须要处理 Flux 或 Mono 中的消息时，能够经过 subscribe 方法来添加相应的订阅逻辑。在调用 subscribe 方法时能够指定须要处理的消息类型。能够只处理其中包含的正常消息，也能够同时处理错误消息和完成消息。

经过 subscribe() 方法处理正常和错误消息

Flux.just(1, 2)
    .concatWith(Mono.error(new IllegalStateException()))
    .subscribe(System.out::println, System.err::println);
复制代码

结果：

1
2
java.lang.IllegalStateException
复制代码

正常的消息处理相对简单。当出现错误时，有多种不一样的处理策略:

经过 onErrorReturn() 方法返回一个默认值

Flux.just(1, 2)
    .concatWith(Mono.error(new IllegalStateException()))
    .onErrorReturn(0)
    .subscribe(System.out::println);
复制代码

结果：

1
2
0
复制代码

经过 onErrorResume()方法来根据不一样的异常类型来选择要使用的产生元素的流

Flux.just(1, 2)
        .concatWith(Mono.error(new IllegalArgumentException()))
        .onErrorResume(e -> {
            if (e instanceof IllegalStateException) {
                return Mono.just(0);
            } else if (e instanceof IllegalArgumentException) {
                return Mono.just(-1);
            }
            return Mono.empty();
            }).subscribe(System.out::println);
复制代码

结果：

1
2
-1
复制代码

经过 retry 操做符来进行重试，重试的动做是经过从新订阅序列来实现的。在使用 retry 操做符时能够指定重试的次数。

Flux.just(1, 2)
    .concatWith(Mono.error(new IllegalStateException()))
    .retry(1)
    .subscribe(System.out::println);
复制代码

结果：

1
2
1
2
Exception in thread "main" reactor.core.Exceptions$ErrorCallbackNotImplemented: java.lang.IllegalStateException
Caused by: java.lang.IllegalStateException
	at com.glmapper.bridge.boot.reactor.SimpleTest.testFluxSub(SimpleTest.java:75)
	at com.glmapper.bridge.boot.reactor.SimpleTest.main(SimpleTest.java:23)
复制代码

调度器 Scheduler

在 Reactor 中，执行模式以及执行过程取决于所使用的 Scheduler，Scheduler 是一个拥有普遍实现类的抽象接口，Schedulers 类提供的静态方法用于达成以下的执行环境：

当前线程（Schedulers.immediate()）

Schedulers.immediate().schedule(()->{
    System.out.println(Thread.currentThread().getName()+"-"+11);
 });
 
 // main-11
复制代码

可重用的单线程（Schedulers.single()）。注意，这个方法对全部调用者都提供同一个线程来使用，直到该调度器（Scheduler）被废弃。若是你想使用专注的线程，就对每个调用使用 Schedulers.newSingle()。
```
Schedulers.single().schedule(()->{
    System.out.println(Thread.currentThread().getName()+"-"+11);
});

// single-1-11
复制代码
```
弹性线程池（Schedulers.elastic()。它根据须要建立一个线程池，重用空闲线程。线程池若是空闲时间过长（默认为 60s）就会被废弃。对于 I/O 阻塞的场景比较适用。 Schedulers.elastic() 可以方便地给一个阻塞的任务分配它本身的线程，从而不会妨碍其余任务和资源。
```
Schedulers.elastic().schedule(()->{
    System.out.println(Thread.currentThread().getName()+"-"+11);
});

// elastic-2-11
复制代码
```

固定大小线程池（Schedulers.parallel()）。所建立线程池的大小与 CPU 个数等同

Schedulers.parallel().schedule(()->{
    System.out.println(Thread.currentThread().getName()+"-"+11);
});

// parallel-1-11
复制代码

基于现有的 ExecutorService 建立 Scheduler

ExecutorService executorService = Executors.newSingleThreadExecutor();
Schedulers.fromExecutorService(executorService).schedule(()->{
    System.out.println(Thread.currentThread().getName()+"-"+11);
});
        
// pool-4-thread-1-11
复制代码

基于 newXXX 方法来建立调度器

Schedulers.newElastic("test-elastic").schedule(()->{
    System.out.println(Thread.currentThread().getName()+"-"+11);
});

// test-elastic-4-11
复制代码

一些操做符默认会使用一个指定的调度器（一般也容许开发者调整为其余调度器）例如，经过工厂方法 Flux.interval(Duration.ofMillis(100)) 生成的每 100ms 打点一次的 Flux，默认状况下使用的是 Schedulers.parallel()，下边的代码演示了如何将其装换为 Schedulers.single()

Flux<String> intervalResult = Flux.interval(Duration.ofMillis(100),
        Schedulers.newSingle("test"))
        .map(i -> Thread.currentThread().getName() +"@"+i);
        intervalResult.subscribe(System.out::println);
复制代码

结果：

test-1@0
test-1@1
test-1@2
test-1@3
test-1@4
// 省略
复制代码

publishOn 和 subscribeOn

Reactor 提供了两种在响应式链中调整调度器 Scheduler 的方法：publishOn 和 subscribeOn。它们都接受一个 Scheduler 做为参数，从而能够改变调度器。可是 publishOn 在链中出现的位置是有讲究的，而 subscribeOn 则无所谓。

publishOn 的用法和处于订阅链（subscriber chain）中的其余操做符同样。它将上游信号传给下游，同时执行指定的调度器 Scheduler 的某个工做线程上的回调。它会改变后续的操做符的执行所在线程（直到下一个 publishOn 出如今这个链上）
subscribeOn 用于订阅（subscription）过程，做用于那个向上的订阅链（发布者在被订阅时才激活，订阅的传递方向是向上游的）。因此，不管你把 subscribeOn 至于操做链的什么位置，它都会影响到源头的线程执行环境（context）。可是，它不会影响到后续的 publishOn，后者仍可以切换其后操做符的线程执行环境。

Flux.create(sink -> {
        sink.next(Thread.currentThread().getName());
        sink.complete();
    })
    .publishOn(Schedulers.single())
    .map(x -> String.format("[%s] %s", Thread.currentThread().getName(), x))
    .publishOn(Schedulers.elastic())
    .map(x -> String.format("[%s] %s", Thread.currentThread().getName(), x))
    .subscribeOn(Schedulers.parallel())
    .toStream()
    .forEach(System.out::println);
复制代码

结果：

[elastic-2] [single-1] parallel-1
复制代码

上面这段代码使用 create() 方法建立一个新的 Flux 对象，其中包含惟一的元素是当前线程的名称。

接着是两对 publishOn() 和 map()方法，其做用是先切换执行时的调度器，再把当前的线程名称做为前缀添加。

最后经过 subscribeOn()方法来改变流产生时的执行方式。

最内层的线程名字 parallel-1 来自产生流中元素时使用的 Schedulers.parallel()调度器，中间的线程名称 single-1 来自第一个 map 操做以前的 Schedulers.single() 调度器，最外层的线程名字 elastic-2 来自第二个 map 操做以前的 Schedulers.elastic()调度器。

先到这里，剩下的想到再补充...