（4）Reactor 3快速上手——响应式Spring的道法术器

Bertha 。 2022-10-28 10:10 230阅读 0赞

boot最新为：2.4.2

里面：spring-boot-starter-webflux 引用的即是，最新的 reactor 3.4.2

<dependency>
      <groupId>io.projectreactor</groupId>
      <artifactId>reactor-core</artifactId>
      <version>3.4.2</version>
      <scope>compile</scope>
    </dependency>

找的路径为：

<artifactId>spring-boot-starter-webflux</artifactId>
      <artifactId>spring-boot-starsster-reactor-netty</artifactId>
      <artifactId>reactor-netty-http</artifactId>
      <artifactId>reactor-core</artifactId>

# （4）Reactor 3快速上手——响应式Spring的道法术器 #

### 1.3.2 Project Reactor ###

[Project Reactor][]（以下简称“Reactor”）与Spring是兄弟项目，侧重于Server端的响应式编程，

*  主要 artifact 是 reactor-core，
 *  这是一个基于 Java 8 的实现了响应式流规范 （Reactive Streams specification）的响应式库。

本文对Reactor的介绍以基本的概念和简单的使用为主，深度以能够满足基本的Spring WebFlux使用为准。

*  在下一章，我会结合Reactor的设计模式、并发调度模型 等原理层面的内容系统介绍Reactor的使用。

光说不练假把式，我们先把练习用的项目搭起来。先创建一个maven项目，然后添加依赖：

<dependency>
            <groupId>io.projectreactor</groupId>
            <artifactId>reactor-core</artifactId>
            <version>3.1.4.RELEASE</version>
        </dependency>

最新版本可到 [http://search.maven.org][http_search.maven.org] 查询，复制过来即可。另外出于测试的需要，添加如下依赖：

<dependency>
            <groupId>io.projectreactor</groupId>
            <artifactId>reactor-test</artifactId>
            <version>3.1.4.RELEASE</version>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.12</version>
            <scope>test</scope>
        </dependency>

好了，我们开始Coding吧。

#### 1.3.2.1 Flux与Mono ####

Reactor中的发布者（Publisher）

*  由`Flux`和`Mono`两个类定义，它们都提供了丰富的操作符（operator）。
 *  一个Flux对象代表一个包含 0…N 个元素的 响应式序列，
 *  而一个Mono对象代表一个 包含 零/一个（0…1）元素的结果。

既然是“数据流”的发布者，Flux和Mono都可以发出三种“数据信号”：

*  元素值、
 *  错误信号、
 *  完成信号，
 *  错误信号和完成信号都是终止信号，
 *  完成信号用于告知 下游订阅者 该数据流 正常结束，
 *  错误信号 终止数据流的同时将 错误传递给 下游订阅者。

下图所示就是一个Flux类型的数据流，黑色箭头是时间轴。

*  它连续发出“1” - “6”共6个元素值，
 *  以及一个完成信号（图中⑥后边的加粗竖线来表示），
 *  完成信号告知订阅者数据流已经结束。

\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-h2rZSlG0-1612524630318)(https://leanote.com/api/file/getImage?fileId=5a90f960ab64414490000a22)\]

下图所示是一个Mono类型的数据流，它发出一个元素值后，又发出一个完成信号。

\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-GmLP9fE1-1612524630320)(https://leanote.com/api/file/getImage?fileId=5a90f9e2ab64414699000ad6)\]

既然Flux具有发布一个数据元素的能力，为什么还要专门定义一个Mono类呢？

*  举个例子，一个HTTP请求产生一个响应，所以对其进行“count”操作是没有多大意义的。
 *  表示这样一个结果的话，应该用`Mono<HttpResponse>`而不是 `Flux<HttpResponse>`，
 *  对于的操作通常只用于处理 0/1 个元素。
 *  它们从语义上就原生包含着元素个数的信息，从而避免了对Mono对象进行多元素场景下的处理。

> 有些操作可以改变基数，从而需要切换类型。比如，count操作用于Flux，
> 
> 但是操作返回的结果是`Mono<Long>`。

我们可以用如下代码声明上边两幅图所示的Flux和Mono：

Flux.just(1, 2, 3, 4, 5, 6);
    Mono.just(1);

Flux和Mono提供了多种创建数据流的方法，

*  `just`就是一种比较直接的 声明数据流 的方式，其参数就是 数据元素。

对于图中的Flux，还可以通过如下方式声明（分别基于数组、集合和Stream生成）：

\`\`\`java  
Integer\[\] array = new Integer\[\]\{1,2,3,4,5,6\};  
Flux.fromArray(array);//基于数组  
List list = Arrays.asList(array);  
Flux.fromIterable(list);//基于集合  
Stream stream = list.stream();  
Flux.fromStream(stream);//基于Stream

不过，这三种信号都不是一定要具备的：
    
    - 首先，错误信号和 完成信号都是终止信号，二者不可能同时共存；
    - 如果没有 发出任何一个元素值，而是 直接发出完成/错误信号，表示这是一个空数据流；
    - 如果没有 错误信号和 完成信号，那么就是一个无限数据流。
    
    比如，对于只有完成/错误信号的数据流：
    
    
    
    ```java
    // 只有完成信号的空数据流
    Flux.just();
    Flux.empty();
    Mono.empty();
    Mono.justOrEmpty(Optional.empty());
    // 只有错误信号的数据流
    Flux.error(new Exception("some error"));
    Mono.error(new Exception("some error"));

你可能会纳闷，空的数据流有什么用？举个例子，当我们从响应式的 DB 中获取结果的时候（假设DAO层是`ReactiveRepository<User>`），就有可能为空：

Mono<User> findById(long id);
     Flux<User> findAll();

无论是空还是发生异常，都需要通过完成/错误信号告知订阅者，已经查询完毕，但是抱歉没有得到值，礼貌问题嘛~

#### 1.3.2.2 订阅前什么都不会发生 ####

数据流有了，假设我们想把每个数据元素原封不动地打印出来：

Flux.just(1, 2, 3, 4, 5, 6).subscribe(System.out::print);
    System.out.println();
    Mono.just(1).subscribe(System.out::println);

输出如下：

123456
    1

可见，`subscribe`方法中的lambda表达式作用在了每一个数据元素上。此外，Flux和Mono还提供了多个`subscribe`方法的变体：

// 订阅并触发数据流
    subscribe(); 
    
    // 订阅并指定对正常数据元素如何处理
    subscribe(Consumer<? super T> consumer); 
    
    // 订阅并定义对正常数据元素和 错误信号的处理
    subscribe(Consumer<? super T> consumer,
              Consumer<? super Throwable> errorConsumer); 
    
    // 订阅并定义对正常数据元素、错误信号和 完成信号的处理
    subscribe(Consumer<? super T> consumer,
              Consumer<? super Throwable> errorConsumer,
              Runnable completeConsumer); 
    
    // 订阅并定义对正常数据元素、错误信号和 完成信号的处理，以及订阅发生时的处理逻辑
    subscribe(Consumer<? super T> consumer,
              Consumer<? super Throwable> errorConsumer,
              Runnable completeConsumer,
              Consumer<? super Subscription> subscriptionConsumer);

1）如果是订阅上边声明的Flux：

Flux.just(1, 2, 3, 4, 5, 6).subscribe(
        System.out::println,
        System.err::println,
        () -> System.out.println("Completed!"));

输出如下：

1
    2
    3
    4
    5
    6
    Completed!

2）再举一个有错误信号的例子：

Mono.error(new Exception("some error")).subscribe(
            System.out::println,
            System.err::println,
            () -> System.out.println("Completed!")
    );

输出如下：

java.lang.Exception: some error  //红字

打印出了错误信号，没有输出`Completed!`表明没有发出完成信号。

这里需要注意的一点是，`Flux.just(1, 2, 3, 4, 5, 6)`仅仅声明了这个数据流，此时数据元素并未发出，只有`subscribe()`方法调用的时候才会触发数据流。所以，**订阅前什么都不会发生**。

#### 1.3.2.3 测试与调试 ####

从命令式 和 同步式编程切换到 响应式和 异步式编程有时候是令人生畏的。学习曲线中最陡峭的地方就是出错时如何分析和调试。

在命令式世界，调试 通常都是非常直观的：

*  直接看 stack trace 就可以找到问题出现的位置， 以及其他信息：是否问题责任全部出在你自己的代码？问题是不是发生在某些库代码？
 *  如果是， 那你的哪部分代码调用了库，是不是传参不合适导致的问题？等等。

当你切换到响应式的异步代码，事情就变得复杂的多了。不过我们先不接触过于复杂的内容，

*  先了解一个基本的单元测试工具——`Step Ve ri fi er`。

最常见的测试 Reactor 序列的场景就是定义一个 Flux 或 Mono，然后在订阅它的时候测试它的行为。

当你的测试关注于每一个数据元素的时候，就非常贴近使用 StepVerifier 的测试场景：

*  下一个期望的数据或信号是什么？
 *  你是否期望使用 Flux 来发出某一个特别的值？
 *  或者是否接下来 300ms 什么都不做？——所有这些都可以使用 StepVerifier API 来表示。

还是以那个1-6的Flux以及会发出错误信号的Mono为例：

private Flux<Integer> generateFluxFrom1To6() { 
            //返回一个 正常的 Flux 流
            return Flux.just(1, 2, 3, 4, 5, 6);
        }
        private Mono<Integer> generateMonoWithError() { 
            //返回一个错误的 Mono流
            return Mono.error(new Exception("some error"));
        }
        @Test
        public void testViaStepVerifier() { 
            //创建 一个流，内容是否是 1-6，执行完毕 核实
            StepVerifier.create(generateFluxFrom1To6())
                    .expectNext(1, 2, 3, 4, 5, 6)
                    .expectComplete()
                    .verify();
            // 创建一个流 是否是 错误的，错误消息是否是 some error
            StepVerifier.create(generateMonoWithError())
                    .expectErrorMessage("some error")
                    .verify();
        }//执行结果为 测试成功

*  其中，`expectNext`用于测试下一个期望的数据元素，
 *  `expectErrorMessage`用于校验下一个元素是否为错误信号，`expectComplete`用于测试下一个元素是否为完成信号。

`StepVerifier`还提供了其他丰富的测试方法，我们会在后续的介绍中陆续接触到。

#### 1.3.2.4 操作符（Operator） ####

*  通常情况下，我们需要对源发布者发出的原始数据流进行多个阶段的处理，并最终得到我们需要的数据。
 *  这种感觉就像是一条流水线，从流水线的源头进入传送带的是原料，经过流水线上各个工位的处理，逐渐由原料变成半成品、零件、组件、成品，最终成为消费者需要的包装品。
 *  这其中，流水线源头的下料机就相当于源发布者，消费者就相当于订阅者，流水线上的一道道工序就相当于一个一个的操作符（Operator）。

下面介绍一些我们常用的操作符。

**1）map - 元素映射为新元素**

`map`操作可以将数据元素进行转换/映射，得到一个新元素。

\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-uAg3XcFc-1612524630322)(https://leanote.com/api/file/getImage?fileId=5a91310eab6441449000118c)\]

public final <V> Flux<V> map(Function<? super T,? extends V> mapper)
    public final <R> Mono<R> map(Function<? super T, ? extends R> mapper)

上图是Flux的map操作示意图，上方的箭头是原始序列的时间轴，下方的箭头是经过map处理后的数据序列时间轴。

`map`接受一个`Function`的 函数式接口为参数，这个函数式的作用是 定义转换操作 的策略。举例说明：

Flux.range(1, 6).map(i -> i * i).subscribe(System.out::println);
    
            StepVerifier.create(Flux.range(1, 6)    // 1
                    .map(i -> i * i))   // 2
                    .expectNext(1, 4, 9, 16, 25, 36)    //3
                    .expectComplete();  // 4
    //有bug,如果把36 改成 37，还是不会报错

1.  `Flux.range(1, 6)` 用于生成从“1”开始的，自增为1的“6”个 整型数据；
2.  `map`接受lambda`i -> i * i`为参数，表示对每个数据进行平方；
3.  验证新的序列的数据；
4.  `verifyComplete()`相当于`expectComplete().verify()`。

**2）flatMap - 元素映射为流**

`flatMap`操作可以将每个数据元素转换/映射为一个流，然后将这些流合并为一个大的数据流。

\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-b6Vv7gi5-1612524630324)(https://leanote.com/api/file/getImage?fileId=5a913829ab64414699001334)\]

注意到，流的合并是异步的，先来先到，并非是严格按照原始序列的顺序（如图蓝色和红色方块是交叉的）。

public final <R> Flux<R> flatMap
        (Function<? super T, ? extends Publisher<? extends R>> mapper)
        
    public final <R> Mono<R> flatMap
        (Function<? super T, ? extends Mono<? extends R>> transformer)

`flatMap`也是接收一个`Function`的函数式接口为参数，

*  这个函数式的输入为一个T类型数据值，对于Flux来说输出可以是Flux和Mono，
 *  对于Mono来说输出只能是Mono。举例说明：

StepVerifier.create(
            Flux.just("flux", "mono")
                    .flatMap(s -> Flux.fromArray(s.split("\\s*"))   // 1
                            .delayElements(Duration.ofMillis(100))) // 2
                    .doOnNext(System.out::print)) // 3
            .expectNextCount(8) // 4
            .verifyComplete();
    //输出的内容为：fmlounox
    //如果 s 换成 a，依然是：mfolunxo，这是正则，现在不懂

1.  对于每一个字符串`s`，将其拆分为包含一个字符的字符串流；
2.  对每个元素延迟100ms；
3.  对每个元素进行打印（注`doOnNext`方法是“偷窥式”的方法，不会消费数据流）；
4.  验证是否发出了8个元素。

打印结果为`mfolnuox`，原因在于各个拆分后的小字符串都是间隔100ms发出的，因此会交叉。

`flatMap`通常用于每个元素又会引入数据流的情况，

*  比如我们有一串url数据流，需要请求每个url并收集response数据。
 *  假设响应式的请求方法如下：

Mono<HttpResponse> requestUrl(String url) { ...}

而url数据流为一个`Flux<String> urlFlux`，那么为了得到所有的HttpResponse，就需要用到flatMap：

urlFlux.flatMap(url -> requestUrl(url));

其返回内容为`Flux<HttpResponse>`类型的HttpResponse流。

**3）filter - 过滤**

`filter`操作可以对数据元素进行筛选。

\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-EpnpEbkG-1612524630325)(https://leanote.com/api/file/getImage?fileId=5a913515ab644146990012ea)\]

public final Flux<T> filter(Predicate<? super T> tester)
    public final Mono<T> filter(Predicate<? super T> tester)

`filter`接受一个`Predicate`的函数式接口为参数，这个函数式的作用是进行判断并返回boolean。举例说明：

StepVerifier.create(Flux.range(1, 6)
                .filter(i -> i % 2 == 1)    // 1
                .map(i -> i * i))
                .expectNext(1, 9, 25)   // 2
                .verifyComplete();
    //过滤过后 为 1 3 5

1.  `filter`的lambda参数表示过滤操作将保留奇数；
2.  验证仅得到奇数的平方。

**4）zip - 一对一合并**

看到`zip`这个词可能会联想到拉链，它能够将多个流一对一的合并起来。

*  zip有多个方法变体，我们介绍一个最常见的二合一的。

zip 
    英 /zɪp/  美 /zɪp/  全球(美国)  
    简明 牛津 新牛津  韦氏  柯林斯 例句  百科
    n. 拉链；活力，精力；尖啸声，撕裂声；一种程序压缩的档案文件格式
    n. (Zip)兹普（人名）
    vi. 拉开或拉上；以尖啸声行进
    vt. 给...以速度；拉上或拉开拉链

\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-eRnDyQb4-1612524630327)(https://leanote.com/api/file/getImage?fileId=5a914413ab64414699001476)\]

它对两个Flux/Mono流每次各取一个元素，合并为一个二元组（`Tuple2`）：

public static <T1,T2> Flux<Tuple2<T1,T2>> zip(Publisher<? extends T1> source1,
                                              Publisher<? extends T2> source2)
    public static <T1, T2> Mono<Tuple2<T1, T2>> zip(Mono<? extends T1> p1, Mono<? extends T2> p2)

*  `Flux`的`zip`方法接受Flux或Mono为参数，
 *  `Mono`的`zip`方法只能接受Mono类型的参数。

举个例子，假设我们有一个关于`zip`方法的说明：“Zip two sources together, that is to say wait for all the sources to emit one element and combine these elements once into a Tuple2.”，我们希望将这句话拆分为一个一个的单词并以每200ms一个的速度发出，除了前面flatMap的例子中用到的`delayElements`，可以如下操作：

private Flux<String> getZipDescFlux() { 
            //定义字符串
            String desc = "Zip two sources together, that is to say wait for all the sources to emit one element and combine these elements once into a Tuple2.";
            //字符串 以空格 剪切
            return Flux.fromArray(desc.split("\\s+"));  // 1
        }
    
        @Test
        public void testSimpleOperators() throws InterruptedException { 
            
            CountDownLatch countDownLatch = new CountDownLatch(1);  // 2
            Flux.zip(
                    getZipDescFlux(),
                    Flux.interval(Duration.ofMillis(200))
                    )  // 3等 200毫秒
    
                    .subscribe(
                            t -> System.out.println(t.getT1()),
                            null,
                            countDownLatch::countDown);    // 4
    
            countDownLatch.await(10, TimeUnit.SECONDS);     // 5
        }

1.  将英文说明用空格拆分为字符串流；
2.  定义一个`CountDownLatch`，初始为1，
    
    1.  则会等待执行1次`countDown`方法后结束，
    2.  不使用它的话，测试方法所在的线程会直接返回 而不会等待数据流发出完毕；
3.  使用`Flux.interval`声明一个每200ms发出一个元素的long数据流；因为zip操作是一对一的，故而将其与字符串流zip之后，字符串流也将具有同样的速度；
4.  zip之后的流中元素类型为`Tuple2`，使用`getT1`方法拿到字符串流的元素；定义完成信号的处理为`countDown`;
5.  `countDownLatch.await(10, TimeUnit.SECONDS)`会等待`countDown`倒数至0，最多等待10秒钟。

除了`zip`静态方法之外，还有`zipWith`等非静态方法，效果与之类似：

getZipDescFlux().zipWith(Flux.interval(Duration.ofMillis(200)))

在异步条件下，数据流的流速不同，使用zip能够一对一地将两个或多个数据流的元素对齐发出。

**5）更多**

Reactor中提供了非常丰富的操作符，除了以上几个常见的，还有：

*  用于编程方式自定义生成数据流的`create`和`generate`等及其变体方法；
 *  用于“无副作用的peek”场景的`doOnNext`、`doOnError`、`doOncomplete`、`doOnSubscribe`、`doOnCancel`等及其变体方法；
 *  用于数据流转换的`when`、`and/or`、`merge`、`concat`、`collect`、`count`、`repeat`等及其变体方法；
 *  用于过滤/拣选的`take`、`first`、`last`、`sample`、`skip`、`limitRequest`等及其变体方法；
 *  用于错误处理的`timeout`、`onErrorReturn`、`onErrorResume`、`doFinally`、`retryWhen`等及其变体方法；
 *  用于分批的`window`、`buffer`、`group`等及其变体方法；
 *  用于线程调度的`publishOn`和`subscribeOn`方法。

使用这些操作符，你几乎可以搭建出能够进行任何业务需求的数据处理管道/流水线。

抱歉以上这些暂时不能一一介绍，更多详情请参考[JavaDoc][]，在下一章我们还会回头对Reactor从更深层次进行系统的分析。

此外，也可阅读我翻译的[Reactor参考文档][Reactor]，我会尽量及时更新翻译的内容。文档源码位于[github][]，如有翻译不当，欢迎提交Pull-Request。

学不动了。

https://blog.51cto.com/liukang/2090191

#### 1.3.2.8 总结 ####

以上关于Reactor的介绍主要是概念层面和使用层面的介绍，不过应该也足以应对常见的业务环境了。

从命令式编程到响应式编程的切换并不是一件容易的事，需要一个适应的过程。不过相信你通过本节的了解和实操，已经可以体会到使用Reactor编程的一些特点：

*  相对于传统的基于回调和Future的异步开发方式，响应式编程更加具有**可编排性和可读性**，配合lambda表达式，代码更加简洁，处理逻辑的表达就像装配“流水线”，适用于对数据流的处理；
 *  在**订阅（subscribe）时才触发数据流**，这种数据流叫做“冷”数据流，就像插座插上电器才会有电流一样，还有一种数据流不管是否有订阅者订阅它都会一直发出数据，称之为“热”数据流，Reactor中几乎都是“冷”数据流；
 *  **调度器对线程管理进行更高层次的抽象**，使得我们可以非常容易地切换线程执行环境；
 *  **灵活的错误处理机制**有利于编写健壮的程序；
 *  **“回压”机制**使得订阅者可以无限接受数据并让它的源头“满负荷”推送所有的数据，也可以通过使用`request`方法来告知源头它一次最多能够处理 n 个元素，从而将“推送”模式转换为“推送+拉取”混合的模式。

后续随着对Reactor的了解我们还会逐渐了解它更多的好玩又好用的特性。

Reactor的开发者中也有来自RxJava的大牛，因此Reactor中甚至许多方法名都是来自RxJava的API的，学习了Reactor之后，很轻松就可以上手Rx家族的库了。

[Project Reactor]: http://projectreactor.io/
[http_search.maven.org]: http://search.maven.org/
[JavaDoc]: http://projectreactor.io/docs/core/release/api/reactor/core/publisher/Flux.html
[Reactor]: https://htmlpreview.github.io/?https://github.com/get-set/reactor-core/blob/master-zh/src/docs/index.html
[github]: https://github.com/get-set/reactor-core/tree/master-zh/src/docs/asciidoc