全球观察：响应式圣经：10W字，实现Spring响应式编程自由

文章很长，而且持续更新，建议收藏起来，慢慢读！疯狂创客圈总目录 博客园版为您奉上珍贵的学习资源 ：

【资料图】

免费赠送 :《尼恩Java面试宝典》持续更新+ 史上最全 + 面试必备 2000页+ 面试必备 + 大厂必备 +涨薪必备免费赠送 经典图书:《Java高并发核心编程（卷1）加强版》面试必备 + 大厂必备 +涨薪必备 加尼恩免费领免费赠送 经典图书:《Java高并发核心编程（卷2）加强版》面试必备 + 大厂必备 +涨薪必备 加尼恩免费领免费赠送 经典图书:《Java高并发核心编程（卷3）加强版》面试必备 + 大厂必备 +涨薪必备 加尼恩免费领

免费赠送 资源宝库： Java 必备 百度网盘资源大合集 价值>10000元 加尼恩领取

前言

全链路异步化改造的基础是响应式编程

随着业务的发展，微服务应用的流量越来越大，使用到的资源也越来越多。

在微服务架构下，大量的应用都是 SpringCloud 分布式架构，这种架构总体上是全链路同步模式。

全链路同步模式不仅造成了资源的极大浪费，并且在流量发生激增波动的时候，受制于系统资源而无法快速的扩容。

全球后疫情时代，降本增效是大背景。如何降本增效？

可以通过技术升级，全链路同步模式，升级为 全链路异步模式。

先回顾一下全链路同步模式架构图

全链路同步模式，如何升级为 全链路异步模式， 就是一个一个 环节的异步化。

40岁老架构师尼恩，持续深化自己的3高架构知识宇宙，当然首先要去完成一次牛逼的全链路异步模式 微服务实操，下面是尼恩的实操过程、效果、压测数据(性能足足提升10倍多)。

全链路异步模式改造 具体的内容，请参考尼恩的深度文章：全链路异步，让你的 SpringCloud 性能优化10倍+

并且，上面的文章，作为尼恩 全链路异步的架构知识，收录在《尼恩Java面试宝典》V46版的架构专题中

全链路异步化改造，性能提升十倍是大好事，那么，全链路同步模式改造的问题是什么呢？

全链路异步化改造的技术基础，是响应式编程，关键问题在于： 响应式编程的知识太难。

古语说： 蜀道难难于上青天。

很多小伙伴认为： 响应式编程， 比蜀道还难？

所以，40岁老架构师 使用20年的编程功力，给大家呈上一篇， 学习响应式编程 的超级长文，也是一篇超级、超级详细，超级超级全面，并且不断迭代的文章：

《响应式圣经：10W字实现响应式编程自由》

此文，目前为V2版本，新版本是基于V1老版本是尼恩之前写的一篇深受好评的博客文章

Flux、Mono、Reactor 实战（史上最全）

后续，尼恩会一直不断迭代， 为大家拉涅出一本史上最棒的 《响应式圣经》，帮助大家实现响应式编程自由。

从此： 蜀道，不再难。

注：本文以 PDF 持续更新，最新尼恩 架构笔记、面试题 的PDF文件，请从这里获取：码云

Reactive programming 响应式编程概述

背景知识

为了应对高并发服务器端开发场景，在2009 年，微软提出了一个更优雅地实现异步编程的方式——Reactive Programming，我们称之为响应式编程。

随后，Netflix 和LightBend 公司提供了RxJava 和Akka Stream 等技术，使得Java 平台也有了能够实现响应式编程的框架。

在2017 年9 月28 日，Spring 5 正式发布。

Spring 5 发布最大的意义在于，它将响应式编程技术的普及向前推进了一大步。

而同时，作为在背后支持Spring 5 响应式编程的框架Spring Reactor，也进入了里程碑式的3.1.0 版本。

什么是响应式编程

响应式编程是一种面向数据流和变化传播的编程范式。

这意味着可以在编程语言中很方便地表达静态或动态的数据流，而相关的计算模型会自动将变化的值通过数据流进行传播。

响应式编程基于reactor（Reactor 是一个运行在 Java8 之上的响应式框架）的思想，当你做一个带有一定延迟的才能够返回的io操作时，不会阻塞，而是立刻返回一个流，并且订阅这个流，当这个流上产生了返回数据，可以立刻得到通知并调用回调函数处理数据。

电子表格程序就是响应式编程的一个例子。有些单元个含有公式，比如可以包含字面值或类似"=B1+C1"的公式，这些单元格的值，会依据其他单元格的值的变化而变化。

响应式传播核心特点之一：变化传播。

一个单元格变化之后，会像多米诺骨牌一样，导致直接和间接引用它的其他单元格均发生相应变化。

从API视角，看什么是 Reactive programming 响应式编程

有了 Reactive Streams 这种标准和规范，利用规范可以进行响应式编程。

那再了解下什么是 Reactive programming 响应式编程。

响应式编程是基于异步和事件驱动的非阻塞程序，只是垂直通过在 JVM 内启动少量线程扩展，而不是水平通过集群扩展。

异步调用 +IO Reactor 事件驱动，可以避免将 CPU 浪费在等待网络 IO 和磁盘 IO 时上，实现提高资源使用率。

Reactive programming就是一个编程范例，具体项目中如何体现呢？

响应式项目编程实战中，通过基于 Reactive Streams 规范实现的框架Spring Reactor 去实战。

Spring Reactor 一般提供两种响应式 API ：

• Mono：实现发布者，并返回 0 或 1 个元素
• Flux：实现发布者，并返回 N 个元素

响应式编程->异步非阻塞

上面讲了响应式编程是什么：

响应式编程（reactive programming）是一种基于数据流（data stream）和变化传递（propagation of change）的声明式（declarative）的编程范式

也讲解了数据流/变化传递/声明式是什么意思，但说到响应式编程就离不开异步非阻塞。

从Spring官网介绍WebFlux的信息我们就可以发现asynchronous, nonblocking这样的字样，因为响应式编程它是异步的，也可以理解成变化传递它是异步执行的。

基于Java8观察者模式

Observable类：此类表示可观察对象，或模型视图范例中的“数据”。

它可以被子类实现以表示应用程序想要观察的对象。

//想要观察的对象 ObserverDemopublic class ObserverDemo extends Observable {    public static void main(String[] args) {        ObserverDemo observerDemo = new ObserverDemo();        //添加观察者        observerDemo.addObserver((o,arg)->{            System.out.println("数据发生变化A");        });        observerDemo.addObserver((o,arg)->{            System.out.println("数据发生变化B");        });        observerDemo.setChanged();//将此Observable对象标记为已更改        observerDemo.notifyObservers();//如果该对象发生了变化，则通知其所有观察者    }}

启动程序测试：

创建一个Observable

rxjava中，可以使用Observable.create() 该方法接收一个Obsubscribe对象

Observable observable = Observable.create(new Observable.OnSubscribe() {    @Override    public void call(Subscriber subscriber) {    }});

来个大点的例子：

Observable observable=Observable.create(new Observable.OnSubscribe() {    @Override    public void call(Subscriber subscriber) {        for(int i=0;i<5;i++){            subscriber.onNext(i);        }        subscriber.onCompleted();    }});//Observable.subscribe(Observer)，Observer订阅了ObservableSubscription subscribe = observable.subscribe(new Observer() {    @Override    public void onCompleted() {        Log.e(TAG, "完成");    }    @Override    public void onError(Throwable e) {        Log.e(TAG, "异常");    }    @Override    public void onNext(Integer integer) {        Log.e(TAG, "接收Obsverable中发射的值：" + integer);    }});

输出：

接收Obsverable中发射的值：0接收Obsverable中发射的值：1接收Obsverable中发射的值：2接收Obsverable中发射的值：3接收Obsverable中发射的值：4

从上面的例子可以看出，在Observer订阅了Observable后，

Observer作为OnSubscribe中call方法的参数传入，从而调用了Observer的相关方法

基于 Reactor 实现

Reactor 是一个运行在 Java8 之上满足 Reactice 规范的响应式框架，它提供了一组响应式风格的 API。

Reactor 有两个核心类： Flux和 Mono，这两个类都实现 Publisher 接口。

• Flux 类似 RxJava 的 Observable，它可以触发零到多个事件，并根据实际情况结束处理或触发错误。
• Mono 最多只触发一个事件，所以可以把 Mono 用于在异步任务完成时发出通知。

Flux 和 Mono 都是数据流的发布者，使用 Flux 和 Mono 都可以发出三种数据信号：元素值，错误信号，完成信号；错误信号和完成信号都代表终止信号，终止信号用于告诉订阅者数据流结束了，错误信号终止数据流同时把错误信息传递给订阅者。

三种信号的特点：

• 错误信号和完成信号都是终止信号，不能共存
• 如果没有发送任何元素值，而是直接发送错误或者完成信号，表示是空数据流
• 如果没有错误信号，也没有完成信号，表示是无限数据流

引入依赖

    org.projectreactor    reactor-core

just 和 subscribe方法

just()：创建Flux序列，并声明指定数据流

subscribe()：订阅Flux序列，只有进行订阅后才回触发数据流，不订阅就什么都不会发生

public class TestReactor {    public static void main(String[] args) {        //just()：创建Flux序列，并声明数据流，        Flux integerFlux = Flux.just(1, 2, 3, 4);//整形        //subscribe()：订阅Flux序列，只有进行订阅后才回触发数据流，不订阅就什么都不会发生        integerFlux.subscribe(System.out::println);                Flux stringFlux = Flux.just("hello", "world");//字符串        stringFlux.subscribe(System.out::println);                //fromArray(),fromIterable()和fromStream()：可以从一个数组、Iterable 对象或Stream 对象中创建Flux序列        Integer[] array = {1,2,3,4};        Flux.fromArray(array).subscribe(System.out::println);                List integers = Arrays.asList(array);        Flux.fromIterable(integers).subscribe(System.out::println);                Stream stream = integers.stream();        Flux.fromStream(stream).subscribe(System.out::println);    }}

启动测试：

Reactive Streams（响应式流）的特点

要搞清楚这两个概念，必须说一下响应流规范。

它是响应式编程的基石。他具有以下特点：

• 响应流必须是无阻塞的。

• 响应流必须是一个数据流。

• 它必须可以异步执行。

• 并且它也应该能够处理背压。

• 即时响应性:

  只要有可能， 系统就会及时地做出响应。 即时响应是可用性和实用性的基石， 而更加重要的是，即时响应意味着可以快速地检测到问题并且有效地对其进行处理。 即时响应的系统专注于提供快速而一致的响应时间， 确立可靠的反馈上限， 以提供一致的服务质量。 这种一致的行为转而将简化错误处理、 建立最终用户的信任并促使用户与系统作进一步的互动。

• 回弹性：

  系统在出现失败时依然保持即时响应性。 这不仅适用于高可用的、 任务关键型系统——任何不具备回弹性的系统都将会在发生失败之后丢失即时响应性。 回弹性是通过复制、 遏制、 隔离以及委托来实现的。 失败的扩散被遏制在了每个组件内部， 与其他组件相互隔离， 从而确保系统某部分的失败不会危及整个系统，并能独立恢复。 每个组件的恢复都被委托给了另一个（外部的）组件， 此外，在必要时可以通过复制来保证高可用性。 （因此）组件的客户端不再承担组件失败的处理。

• 弹性：

  系统在不断变化的工作负载之下依然保持即时响应性。 反应式系统可以对输入（负载）的速率变化做出反应，比如通过增加或者减少被分配用于服务这些输入（负载）的资源。 这意味着设计上并没有争用点和中央瓶颈， 得以进行组件的分片或者复制， 并在它们之间分布输入（负载）。 通过提供相关的实时性能指标， 反应式系统能支持预测式以及反应式的伸缩算法。 这些系统可以在常规的硬件以及软件平台上实现成本高效的弹性。

• 消息驱动：

  反应式系统依赖异步的、消息传递，从而确保了松耦合、隔离、位置透明的组件之间有着明确边界。

  这一边界还提供了将失败作为消息委托出去的手段。

  使用显式的消息传递，可以通过在系统中塑造并监视消息流队列， 并在必要时应用回压， 从而实现负载管理、 弹性以及流量控制。

  使用位置透明的消息传递作为通信的手段， 使得跨集群或者在单个主机中使用相同的结构成分和语义来管理失败成为了可能。

  非阻塞的通信使得接收者可以只在活动时才消耗资源， 从而减少系统开销。

Reactive Streams（响应式流）的过程

一般由以下组成：

一般由以下组成：

• publisher：发布者，发布元素到订阅者
• subscriber：订阅者，消费元素
• subscription：订阅，在发布者中，订阅被创建时，将与订阅者共享
• processor：处理器，发布者与订阅者之间处理数据，包含了发布者与订阅者的共同体

publisher接口规范

public interface Publisher {    void subscribe(Subscriber var1); //添加订阅者}

subscriber接口规范

public interface Subscriber {     void onSubscribe(Subscription var1);        void onNext(T var1);        void onError(Throwable var1);      void onComplete();}

subscription接口规范

public interface Subscription {       void request(long var1);       void cancel();}

processor接口规范

public interface Processor extends Subscriber, Publisher {}

Reactor 流框架的组件和使用

• Reactor 框架是 Pivotal 公司（开发 Spring 等技术的公司）开发的
• 实现了 Reactive Programming 思想，符合Reactive Streams 规范（Reactive Streams 是由 Netflix、TypeSafe、Pivotal 等公司发起的）的一项技术
• 侧重于server端的响应式编程框架
• Reactor 框架主要有两个主要的模块：reactor-core 和 reactor-ipc。前者主要负责 Reactive Programming 相关的核心 API 的实现，后者负责高性能网络通信的实现，目前是基于 Netty 实现的。

Java原有的异步编程方式

• Callback：异步方法采用一个callback作为参数，当结果出来后回调这个callback，例如swings的EventListener
• Future：异步方法返回一个Future，此时结果并不是立刻可以拿到，需要处理结束之后才可以使用

Future局限

• 多个Future组合不易
• 调用Future#get时仍然会阻塞
• 缺乏对多个值以及进一步的出错处理

Reactor的Publisher

• Mono 实现了 org.reactivestreams.Publisher 接口，代表0到1个元素的响应式序列。

• Flux 同样实现了 org.reactivestreams.Publisher 接口，代表0到N个元素的结果。

Publisher/Flux和Mono 三大流

由于响应流的特点，我们不能再返回一个简单的POJO对象来表示结果了。

必须返回一个类似Java中的Future的概念，在有结果可用时通知消费者进行消费响应。

Reactive Stream规范中这种被定义为Publisher

Publisher 发射着基础类

Publisher是一个可以提供0-N个序列元素的提供者，并根据其订阅者Subscriber的需求推送元素。

一个Publisher可以支持多个订阅者，并可以根据订阅者的逻辑进行推送序列元素。

下面这个Excel计算就能说明一些Publisher的特点。

A1-A9就可以看做Publisher及其提供的元素序列。

A10-A13分别是求和函数SUM(A1:A9)、平均函数AVERAGE(A1:A9)、最大值函数MAX(A1:A9)、最小值函数MIN(A1:A9)，

A10-A13可以看作订阅者Subscriber。

假如说我们没有A10-A13，那么A1-A9就没有实际意义，它们并不产生 计算。

这也是响应式的一个重要特点：当没有订阅时发布者什么也不做。

而Flux和Mono都是Publisher在Reactor 3实现类。

Publisher提供了subscribe方法，允许增加订阅的消费者，订阅之后，消费者在有结果可用时进行消费。

如果没有消费者，Publisher不会做任何事情，所以，Publisher根据消费情况进行响应。

Publisher可能返回零或者多个，甚至可能是无限的，为了更加清晰表示期待的结果就引入了两个实现模型Mono和Flux。

深入Publishers 发布者（发射序列）

异步处理将I/O或计算与调用该操作的线程进行解耦。

异步处理返回结果的未来句柄，通常是java.util.concurrent.Future或类似的东西，它返回单个对象，集合或异常。

使用Java 8或Promise模式，可以设置future的线性链接，以便发出后续的异步请求。 一旦需要条件处理，就必须中断并同步异步流。 尽管这种方法是可行的，但它并未充分利用异步处理的优势。

Publisher支持值甚至是无限流的发射序列，而不仅是单个标量值的发射（如Future那样）。 一旦开始处理流而不是单个值，你将非常感谢这个事实。

Project Reactor的词汇表使用两种类型：Mono和Flux，它们都是发布者。

Mono可以发出0到1个事件，而Flux可以发出0到N个事件。

Publisher不会偏向某些特定的并发性或异步性来源，也不会偏向于在ThreadPool中运行基础代码的执行方式-同步还是异步。 作为Publisher的消费者，你将实际的实现留给了生产者，生产者可以在以后修改它而无需修改代码。

Publisher的最后一个关键点是，底层处理不是在获取Publisher时开始的，而是在观察者订阅或向 Publisher发出信号的那一刻开始的。

这与java.util.concurrent.Future至关重要，后者在创建/获取（created/obtained）时在某个地方启动。 因此，如果没有观察者订阅Publisher，则将不会发生任何事情。

Publisher和 Iterable (pull)的对比

Publisher 如何使用

与发布者合作时，你要做的第一件事就是消费它们。

消费发布者意味着订阅它。

这是一个订阅并打印所有发出的项目的示例：

Flux.just("Ben", "Michael", "Mark").subscribe(new Subscriber() {    public void onSubscribe(Subscription s) {        s.request(3);    }     public void onNext(String s) {        System.out.println("Hello " + s + "!");    }     public void onError(Throwable t) {     }     public void onComplete() {        System.out.println("Completed");    }});

该示例打印以下行：

Hello BenHello MichaelHello MarkCompleted

你可以看到订阅者（或观察者）收到每个事件的通知，并且还接收到已完成的事件。

Publisher会发出项目（items），直到引发异常或Publisher完成调用onCompleted的发出为止。 在那之后不再发出其他元素。

对subscribe的调用会注册一个允许取消的Subscription，因此不会接收其他事件。

一旦订阅者从Publisher中取消订阅，发布者便可以与取消订阅和免费资源进行互操作。

实现Subscriber需要实现多种方法，因此让我们将代码重写为更简单的形式：

Flux.just("Ben", "Michael", "Mark").doOnNext(new Consumer() {    public void accept(String s) {        System.out.println("Hello " + s + "!");    }}).doOnComplete(new Runnable() {    public void run() {        System.out.println("Completed");    }}).subscribe();

或者，使用Java 8 Lambdas甚至更简单：

Flux.just("Ben", "Michael", "Mark")        .doOnNext(s -> System.out.println("Hello " + s + "!"))        .doOnComplete(() -> System.out.println("Completed"))        .subscribe();

你可以使用运算符控制Subscriber处理的元素。

如果仅对前N个元素感兴趣，take()运算符将限制发射项目的数量。

Flux.just("Ben", "Michael", "Mark") //        .doOnNext(s -> System.out.println("Hello " + s + "!"))        .doOnComplete(() -> System.out.println("Completed"))        .take(2)        .subscribe();

该示例打印以下行：

Hello BenHello MichaelCompleted

请注意，一旦发出预期的元素计数，take操作符就会从Publisher隐式取消其订阅。

可以通过另一个Flux或Subscriber来完成对Publisher的订阅。

除非要实现自定义Publisher，否则请始终使用Subscriber。

上例中使用的订阅者Consumer不处理异常，因此一旦引发异常，你将看到如下堆栈跟踪：

Exception in thread "main" reactor.core.Exceptions$BubblingException: java.lang.RuntimeException: Example exception    at reactor.core.Exceptions.bubble(Exceptions.java:96)    at reactor.core.publisher.Operators.onErrorDropped(Operators.java:296)    at reactor.core.publisher.LambdaSubscriber.onError(LambdaSubscriber.java:117)    ...Caused by: java.lang.RuntimeException: Example exception    at demos.lambda$example3Lambda$4(demos.java:87)    at reactor.core.publisher.FluxPeekFuseable$PeekFuseableSubscriber.onNext(FluxPeekFuseable.java:157)    ... 23 more

始终建议从一开始就实施错误处理程序。 在某些时候，事情可能并且会出错。

完全实现的订阅者声明onCompleted和onError方法，使你可以对以下事件作出反应：

Flux.just("Ben", "Michael", "Mark").subscribe(new Subscriber() {    public void onSubscribe(Subscription s) {        s.request(3);    }     public void onNext(String s) {        System.out.println("Hello " + s + "!");    }     public void onError(Throwable t) {        System.out.println("onError: " + e);    }     public void onComplete() {        System.out.println("Completed");    }});

从push到pull

上面的示例说明了如何以一种非阻塞式或非阻塞式执行的方式设置发布者。

Flux可以显式转换为Iterable或与block()同步。

开始表达代码内部执行的本质时，请避免在代码中调用block()。

调用block()消除了应用程序反应链（reactive chain）的所有非阻塞优势。

String last = Flux.just("Ben", "Michael", "Mark").last().block();System.out.println(last);

该示例打印以下行：

Mark

阻塞调用可用于同步发布者链（chain），并找到进入普通且众所周知的Pull模式的方法。

List list = Flux.just("Ben", "Michael", "Mark").collectList().block();System.out.println(list);

toList运算符收集所有发出的元素，并将列表通过BlockingPublisher传递。

该示例打印以下行：

[Ben, Michael, Mark]

Flux 流（多个元素的发射序列）

Flux 是一个发出(emit)0-N个元素组成的异步序列的Publisher,可以被onComplete信号或者onError信号所终止。

在响应流规范中存在三种给下游消费者调用的方法 onNext, onComplete, 和onError。

下面这张图表示了Flux的抽象模型：

以上的的讲解对于初次接触反应式编程的依然是难以理解的，所以这里有一个循序渐进的理解过程。

有些类比并不是很妥当，但是对于你循序渐进的理解这些新概念还是有帮助的。

传统数据处理

我们在平常是这么写的：

public List allUsers() {    return Arrays.asList(new User("A"),new User("B"));}

我们通过迭代返回值List来get这些元素进行再处理（消费），不管有没有消费者， 菜品都会生产出来。

流式数据处理

在Java 8中我们可以改写为流的表示：

public Stream allUsers() {    return  Stream.of(new User("A"),new User("B"));}

反应式数据处理

在Reactor中我们又可以改写为Flux表示：

public Flux allUsers(){    return Flux.just(new User("A"),new User("B"));}

这时候食客来了，发生了订阅，厨师才开始做。

Flux 的创建Demo

Flux ints = Flux.range(1, 4);Flux seq1 = Flux.just("bole1", "bole2", "bole3");List iterable = Arrays.asList("bole_01", "bole_02", "bole_03");Flux seq2 = Flux.fromIterable(iterable);seq2.subscribe(i -> System.out.println(i));

Mono

Mono 是一个发出(emit)0-1个元素的Publisher,可以被onComplete信号或者onError信号所终止。

mono 整体和Flux差不多，只不过这里只会发出0-1个元素。也就是说不是有就是没有。

象Flux一样，我们来看看Mono的演化过程以帮助理解。

传统数据处理

public User currentUser () {    return isAuthenticated ? new User("felord.cn", "reactive") : null;}

直接返回符合条件的对象或者null`。

Optional的处理方式

public Optional currentUser () {    return isAuthenticated ? Optional.of(new User("felord.cn", "reactive"))            : Optional.empty();}

这个Optional我觉得就有反应式的那种味儿了，当然它并不是反应式。当我们不从返回值Optional取其中具体的对象时，我们不清楚里面到底有没有，但是Optional是一定客观存在的,不会出现NPE问题。

反应式数据处理

public Mono currentUser () {    return isAuthenticated ? Mono.just(new User("felord.cn", "reactive"))            : Mono.empty();}

和Optional有点类似的机制，当然Mono不是为了解决NPE问题的，它是为了处理响应流中单个值（也可能是Void）而存在的。

Mono的创建Demo

Mono data = Mono.just("bole");Mono noData = Mono.empty();m.subscribe(i -> System.out.println(i));

函数编程

反应式编程，常常和函数式编程结合，这就是让大家困扰的地方

函数编程接口

常用函数编程示例

Consumer ：有 一个 输入 无输出 函数接口

Product product=new Product();//类名+静态方法  有一个输入T， 没有输出Consumer consumer1 = Product->Product.nameOf(product);//lambdaconsumer1.accept(product);Consumer consumer = Product::nameOf;//方法引用consumer.accept(product);

Funtion ：一个输入 一个输出 函数接口

//对象+方法  一个输入T 一个输出RFunction function = product::reduceStock;System.out.println("剩余库存：" + function.apply(10));//带参数的构造函数Function function1=Product::new;System.out.println("新对象:" +function1.apply(200));

Predicate ： 一个输入T, 一个输出 Boolean

//Predicate 一个输入T 一个输出BooleanPredicate predicate= i -> product.isEnough(i);//lambdaSystem.out.println("库存是否足够："+predicate.test(100));Predicate predicate1= product::isEnough;//方法引用System.out.println("库存是否足够："+predicate1.test(100));

UnaryOperator ：一元操作符 ,输入输出都是T

//一元操作符  输入和输出TUnaryOperator integerUnaryOperator =product::reduceStock;System.out.println("剩余库存：" + integerUnaryOperator.apply(20));IntUnaryOperator intUnaryOperator = product::reduceStock;System.out.println("剩余库存：" + intUnaryOperator.applyAsInt(30));

Supplier ： 没有输入 只有输出

//无参数构造函数Supplier supplier = Product::new;System.out.println("创建新对象:" + supplier.get());

Supplier supplier1=()->product.getStock();System.out.println("剩余库存:" + supplier1.get());

BiFunction： 二元操作符 两个输入 一个输出

//类名+方法BiFunction binaryOperator = Product::reduceStock;System.out.println(" 剩余库存(BiFunction)：" + binaryOperator.apply(product, 10));

BinaryOperator ：二元操作符，二个输入，一个输出

//BinaryOperator binaryOperator1=(x,y)->product.reduceStock(x,y);BinaryOperator binaryOperator1=product::reduceStock;System.out.println(" 剩余库存(BinaryOperator)：" +binaryOperator1.apply(product.getStock(),10));

Flux类中的静态方法：

简单的创建方法

just()：

可以指定序列中包含的全部元素。创建出来的Flux序列在发布这些元素之后会自动结束

fromArray()，fromIterable()，fromStream()：

可以从一个数组，Iterable对象或Stream对象中穿件Flux对象

empty()：

创建一个不包含任何元素，只发布结束消息的序列

error(Throwable error)：

创建一个只包含错误消息的序列

never()：

传建一个不包含任务消息通知的序列

range(int start, int count)：

创建包含从start起始的count个数量的Integer对象的序列

interval(Duration period)和interval(Duration delay, Duration period)：

创建一个包含了从0开始递增的Long对象的序列。其中包含的元素按照指定的间隔来发布。除了间隔时间之外，还可以指定起始元素发布之前的延迟时间

intervalMillis(long period)和intervalMillis(long delay, long period)：

与interval()方法相同，但该方法通过毫秒数来指定时间间隔和延迟时间

例子

Flux.just("Hello", "World").subscribe(System.out::println);Flux.fromArray(new Integer[]{1, 2, 3}).subscribe(System.out::println);Flux.empty().subscribe(System.out::println);Flux.range(1, 10).subscribe(System.out::println);Flux.interval(Duration.of(10, ChronoUnit.SECONDS)).subscribe(System.out::println);Flux.intervalMillis(1000).subscirbe(System.out::println);

复杂的序列创建 generate()

当序列的生成需要复杂的逻辑时，则应该使用generate()或create()方法。

generate()方法通过同步和逐一的方式来产生Flux序列。

序列的产生是通过调用所提供的的SynchronousSink对象的next()，complete()和error(Throwable)方法来完成的。

逐一生成的含义是在具体的生成逻辑中，next()方法只能最多被调用一次。

在某些情况下，序列的生成可能是有状态的，需要用到某些状态对象，此时可以使用

generate(Callable stateSupplier, BiFunction, S> generator)，

其中stateSupplier用来提供初始的状态对象。

在进行序列生成时，状态对象会作为generator使用的第一个参数传入，可以在对应的逻辑中对改状态对象进行修改以供下一次生成时使用。

Flux.generate(sink -> {    sink.next("Hello");    sink.complete();  }).subscribe(System.out::println);final Random random = new Random();Flux.generate(ArrayList::new, (list, sink) -> {    int value = random.nextInt(100);    list.add(value);    sink.next(value);    if( list.size() ==10 )        sink.complete();    return list;}).subscribe(System.out::println);

复杂的序列创建 create()

create()方法与generate()方法的不同之处在于所使用的是FluxSink对象。

FluxSink支持同步和异步的消息产生，并且可以在一次调用中产生多个元素。

Flux.create(sink -> {    for(int i = 0; i < 10; i ++)        sink.next(i);    sink.complete();}).subscribe(System.out::println);

Mono静态方法

Mono类包含了与Flux类中相同的静态方法：just()，empty()和never()等。

除此之外，Mono还有一些独有的静态方法：

• fromCallable()，fromCompletionStage()，fromFuture()，fromRunnable()和fromSupplier()：分别从Callable，CompletionStage，CompletableFuture，Runnable和Supplier中创建Mono

• delay(Duration duration)和delayMillis(long duration)：创建一个Mono序列，在指定的延迟时间之后，产生数字0作为唯一值

• ignoreElements(Publisher source)：创建一个Mono序列，忽略作为源的Publisher中的所有元素，只产生消息

• justOrEmpty(Optional data)和justOrEmpty(T data)：从一个Optional对象或可能为null的对象中创建Mono。只有Optional对象中包含之或对象不为null时，Mono序列才产生对应的元素

例子：

Mono.fromSupplier(() -> "Hello").subscribe(System.out::println);Mono.justOrEmpty(Optional.of("Hello")).subscribe(System.out::println);Mono.create(sink -> sink.success("Hello")).subscribte(System.out::println);

操作符

操作符buffer和bufferTimeout

这两个操作符的作用是把当前流中的元素收集到集合中，并把集合对象作为流中的新元素。

Flux.range(1, 100).buffer(20).subscribe(System.out::println);

在进行收集时可以指定不同的条件：所包含的元素的最大数量或收集的时间间隔。

方法buffer()仅使用一个条件，而bufferTimeout()可以同时指定两个条件。

指定时间间隔时可以使用Duration对象或毫秒数，即使用bufferMillis()或bufferTimeoutMillis()两个方法。

除了元素数量和时间间隔外，还可以通过bufferUntil和bufferWhile操作符来进行收集。这两个操作符的参数时表示每个集合中的元素索要满足的条件的Predicate对象。

bufferUntil会一直收集直到Predicate返回true。

使得Predicate返回true的那个元素可以选择添加到当前集合或下一个集合中；bufferWhile则只有当Predicate返回true时才会收集。一旦为false，会立即开始下一次收集。

Flux.intervalMillis(100).bufferMillis(1001).take(2).toStream().forEach(System.out::println);Flux.range(1, 10).bufferUntil(i -> i%2 == 0).subscribe(System.out::println);Flux.range(1, 10).bufferWhile(i -> i%2 == 0).subscribe(System.out::println);

操作符Filter

对流中包含的元素进行过滤，只留下满足Predicate指定条件的元素。

Flux.range(1, 10).filter(i -> i%2 == 0).subscribe(System.out::println);

操作符zipWith

zipWith操作符把当前流中的元素与另一个流中的元素按照一对一的方式进行合并。在合并时可以不做任何处理，由此得到的是一个元素类型为Tuple2的流；也可以通过一个BiFunction函数对合并的元素进行处理，所得到的流的元素类型为该函数的返回值。

Flux.just("a", "b").zipWith(Flux.just("c", "d")).subscribe(System.out::println);Flux.just("a", "b").zipWith(Flux.just("c", "d"), (s1, s2) -> String.format("%s-%s", s1, s2)).subscribe(System.out::println);

操作符take

take系列操作符用来从当前流中提取元素。提取方式如下：

take(long n)，take(Duration timespan)和takeMillis(long timespan)：按照指定的数量或时间间隔来提取

takeLast(long n)：提取流中的最后N个元素

takeUntil(Predicate predicate) ：提取元素直到Predicate返回true

takeWhile(Predicate continuePredicate)：当Predicate返回true时才进行提取

takeUntilOther(Publisher other)：提取元素知道另外一个流开始产生元素

Flux.range(1, 1000).take(10).subscribe(System.out::println);Flux.range(1, 1000).takeLast(10).subscribe(System.out::println);Flux.range(1, 1000).takeWhile(i -> i < 10).subscribe(System.out::println);Flux.range(1, 1000).takeUntil(i -> i == 10).subscribe(System.out::println);

操作符reduce和reduceWith

reduce和reduceWith操作符对流中包含的所有元素进行累计操作，得到一个包含计算结果的Mono序列。累计操作是通过一个BiFunction来表示的。在操作时可以指定一个初始值。若没有初始值，则序列的第一个元素作为初始值。

Flux.range(1, 100).reduce((x, y) -> x + y).subscribe(System.out::println);Flux.range(1, 100).reduceWith(() -> 100, (x + y) -> x + y).subscribe(System.out::println);

操作符merge和mergeSequential

merge和mergeSequential操作符用来把多个流合并成一个Flux序列。merge按照所有流中元素的实际产生序列来合并，而mergeSequential按照所有流被订阅的顺序，以流为单位进行合并。

Flux.merge(Flux.intervalMillis(0, 100).take(5), Flux.intervalMillis(50, 100).take(5)).toStream().forEach(System.out::println);Flux.mergeSequential(Flux.intervalMillis(0, 100).take(5), Flux.intervalMillis(50, 100).take(5)).toStream().forEach(System.out::println);

操作符flatMap和flatMapSequential

flatMap和flatMapSequential操作符把流中的每个元素转换成一个流，再把所有流中的元素进行合并。flatMapSequential和flatMap之间的区别与mergeSequential和merge是一样的。

Flux.just(5, 10).flatMap(x -> Flux.intervalMillis(x * 10, 100).take(x)).toStream().forEach(System.out::println);

操作符concatMap

concatMap操作符的作用也是把流中的每个元素转换成一个流，再把所有流进行合并。concatMap会根据原始流中的元素顺序依次把转换之后的流进行合并，并且concatMap堆转换之后的流的订阅是动态进行的，而flatMapSequential在合并之前就已经订阅了所有的流。

Flux.just(5, 10).concatMap(x -> Flux.intervalMillis(x * 10, 100).take(x)).toStream().forEach(System.out::println);

操作符combineLatest

combineLatest操作符把所有流中的最新产生的元素合并成一个新的元素，作为返回结果流中的元素。只要其中任何一个流中产生了新的元素，合并操作就会被执行一次，结果流中就会产生新的元素。

Flux.combineLatest(Arrays::toString, Flux.intervalMillis(100).take(5), Flux.intervalMillis(50, 100).take(5)).toStream().forEach(System.out::println);

消息处理

当需要处理Flux或Mono中的消息时，可以通过subscribe方法来添加相应的订阅逻辑。

在调用subscribe方法时可以指定需要处理的消息类型。

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

第2种可以通过switchOnError()方法来使用另外的流来产生元素。

Flux.just(1, 2).concatWith(Mono.error(new IllegalStateException())).switchOnError(Mono.just(0)).subscribe(System.out::println);

第三种是通过onErrorResumeWith()方法来根据不同的异常类型来选择要使用的产生元素的流。

Flux.just(1, 2).concatWith(Mono.error(new IllegalArgumentException())).onErrorResumeWith(e -> {    if(e instanceof IllegalStateException)        return Mono.just(0);    else if(e instanceof IllegalArgumentException)        return Mono.just(-1);    return Mono.epmty();}).subscribe(System,.out::println);

当出现错误时还可以使用retry操作符来进行重试。重试的动作是通过重新订阅序列来实现的。在使用retry操作时还可以指定重试的次数。

Flux.just(1, 2).concatWith(Mono.error(new IllegalStateException())).retry(1).subscrible(System.out::println);

测试

StepVerifier的作用是可以对序列中包含的元素进行逐一验证。通过StepVerifier.create()方法对一个流进行包装之后再进行验证。expectNext()方法用来声明测试时所期待的流中的下一个元素的值，而verifyComplete()方法则验证流是否正常结束。verifyError()来验证流由于错误而终止。

StepVerifier.create(Flux.just(a, b)).expectNext("a").expectNext("b").verifyComplete();

使用StepVerifier.withVirtualTime()方法可以创建出使用虚拟时钟的SteoVerifier。通过thenAwait(Duration)方法可以让虚拟时钟前进。

StepVerifier.withVirtualTime(() -> Flux.interval(Duration.ofHours(4), Duration.ofDays(1)).take(2))    .expectSubscription()    .expectNoEvent(Duration.ofHours(4))    .expectNext(0L)    .thenAwait(Duration.ofDays(1))    .expectNext(1L)    .verifyComplete();

TestPublisher的作用在于可以控制流中元素的产生，甚至是违反反应流规范的情况。通过create()方法创建一个新的TestPublisher对象，然后使用next()方法来产生元素，使用complete()方法来结束流。

final TestPublisher testPublisher = TestPublisher.creater();testPublisher.next("a");testPublisher.next("b");testPublisher.complete();StepVerifier.create(testPublisher)    .expectNext("a")    .expectNext("b")    .expectComplete();

调试

在调试模式启用之后，所有的操作符在执行时都会保存额外的与执行链相关的信息。当出现错误时，这些信息会被作为异常堆栈信息的一部分输出。

Hooks.onOperator(providedHook -> providedHook.operatorStacktrace());

也可以通过checkpoint操作符来对特定的流处理链来启用调试模式。

Flux.just(1, 0).map(x -> 1/x).checkpoint("test").subscribe(System.out::println);

日志记录

可以通过添加log操作把流相关的事件记录在日志中，

Flux.range(1, 2).log("Range").subscribe(System.out::println);

冷热序列

冷序列的含义是不论订阅者在何时订阅该序列，总是能收到序列中产生的全部消息。

热序列是在持续不断的产生消息，订阅者只能获取到在其订阅之后产生的消息。　　

final Flux source = Flux.intervalMillis(1000).take(10).publish.autoConnect();source.subscribe();Thread.sleep(5000);source.toStream().forEach(System.out::println);

map 、flatMap 以及 flatMapSequential 区别：

//Map 的方法签名 Flux map(Function mapper)//FlatMap的方法签名 Flux flatMap(Function> mapper)

map：

通过对每个元素应用 转换函数来 同步 转换此Flux发出的元素， 并且是一对一的转换流元素。

flatMap：

将此Flux发出的元素 异步 转换为 Publisher，然后通过合并将这些内部Publisher，最终扁平化为单个Flux 。

Flux可能包含N个元素，所以flatMap是一对多的转换。

Flux 将各个 publisher 合并的过程中，不会保持 源Flux发布 的顺序，可能出现交错。

flatMapSequential：

将此Flux发出的元素异步转换为 Publisher，然后通过合并将这些内部发布者扁平化为单个Flux 。

与 flatMap 不同的是 flatMapSequential 在合并 publisher 时， 会 按源元素的顺序合并它们。

map是一个同步运算符，它只是一种将一个值转换为另一个值的方法。

flatMap可以是同步的，也可以是异步的，这取决于flatMap中调用的方法怎么使用。

示例1：

void demo() {    //1. Flux.interval 按时生产Long的无限流    final Flux flux = Flux.interval(Duration.of(100, ChronoUnit.MILLIS))            .map(log()) //2. 调用方法log，订阅后log()会在当前线程执行            .flatMap(logOfFlux()); //3. 通过flatMap调用 log(), 会在当前线程执行    //完成flux的定义，调用subscribe后才会真正开始执行    flux.subscribe();}Function log() {    return aLong -> {        log.info("num is {}", aLong);        return aLong;    };}Function> logOfFlux() {    return aLong -> {        log.info("num is {}", aLong);        return Flux.just(aLong);    };}

在我们的示例代码中，该 flatMap 操作是同步的，因为我们使用Flux.just()方法发出元素。

下面我们会介绍如何在 flatMap 中实现异步操作。

上边代码中讲到了 Publisher 调用 subscribe 后才会真正开始执行，所以 subscribe 中的代码并不一定会执行。

当 Mono 是空序列时 ：

void monoOfEmpty() {    Mono.empty()        .map(m -> func())        .subscribe(message -> {          responseObserver.onNext(message);          responseObserver.onCompleted();        );}

会有同学喜欢在 subscribe 中处理响应（例如rpc），但这个场景中responseObserver.onCompleted() 不会被执行。

正确的做法应该是：

void monoOfEmpty() {    Mono.empty()        .map(m -> func())        .doOnSuccess(message -> responseObserver.onCompleted())        .subscribe(responseObserver::onNext);}

如何选择操作符？

本节的内容来自 Reacto 3 参考文档——如何选择操作符。

如果一个操作符是专属于 Flux或 Mono的，那么会给它注明前缀。公共的操作符没有前缀。如果一个具体的用例涉及多个操作符的组合，这里以方法调用的方式展现，会以一个点（.）开头，并将参数置于圆括号内，比如： .methodCall(parameter)。

1）创建一个新序列，它…

1.1 发出一个T，用just：

Mono ,是指最多只能触发(emit) (事件)一次。

Mono 对应于 RxJava 库的 Single 和 Maybe 类型或者是java的Optional。

因此一个异步任务，如果只是想要在完成时给出完成信号，就可以使用 Mono。

@Testpublic void testMonoBasic(){    Mono.fromSupplier(() -> "Hello").subscribe(System.out::println);    Mono.justOrEmpty(Optional.of("Hello")).subscribe(System.out::println);    Mono.create(sink -> sink.success("Hello")).subscribe(System.out::println);}

• 基于一个 Optional：Mono#justOrEmpty(Optional)
• 基于一个可能为 null的 T：Mono#justOrEmpty(T)

Flux 相当于一个 RxJava Observable，能够发出 0~N 个数据项，然后（可选地）completing 或 erroring。Flux处理多个数据项作为stream。

@Testpublic void testBasic(){    Flux.just("Hello", "World").subscribe(System.out::println);    Flux.fromArray(new Integer[] {1, 2, 3}).subscribe(System.out::println);    Flux.empty().subscribe(System.out::println);    Flux.range(1, 10).subscribe(System.out::println);    Flux.interval(Duration.of(10, ChronoUnit.SECONDS)).subscribe(System.out::println);}    

1.2 发出一个“懒”创建的T

• 还是由 just返回但是“懒”创建的：

使用 Mono#fromSupplier或用 defer包装 just

@Testpublic void defer(){    //声明阶段创建DeferClass对象    Mono m1 = Mono.just(new Date());    Mono m2 = Mono.defer(()->Mono.just(new Date()));    m1.subscribe(System.out::println);    m2.subscribe(System.out::println);    //延迟5秒钟    try {        Thread.sleep(5000);    } catch (InterruptedException e) {        e.printStackTrace();    }    m1.subscribe(System.out::println);    m2.subscribe(System.out::println);}

1.3发出许多 T，

• 这些元素我可以明确列举出来：Flux#just(T...)
• 基于迭代数据结构:
  • 一个数组：Flux#fromArray
  • 一个集合或 iterable：Flux#fromIterable
  • 一个 Integer 的 range：Flux#range
  • 一个 Stream提供给每一个订阅：Flux#fromStream(Supplier)
• 基于一个参数值给出的源：
  • 一个 Supplier：Mono#fromSupplier
  • 一个任务：Mono#fromCallable，Mono#fromRunnable
  • 一个 CompletableFuture：Mono#fromFuture
• 直接完成：empty
• 立即生成错误：error
  • …“懒”的方式生成 Throwable：error(Supplier)
• 什么都不做：never
• 订阅时才决定：defer
• 依赖一个可回收的资源：using
• 可编程地生成事件（可以使用状态）:
  • 同步且逐个的：Flux#generate
  • 异步（也可同步）的，每次尽可能多发出元素：Flux#create（Mono#create也类似的，只不过只能发一个）

2）对序列进行转化

• 我想转化一个序列：
  • 1对1地转化（比如字符串转化为它的长度）：map
  • …类型转化：cast
  • …为了获得每个元素的序号：Flux#index
  • 1对n地转化（如字符串转化为一串字符）：flatMap+ 使用一个工厂方法
  • 1对n地转化可自定义转化方法和/或状态：handle
  • 对每一个元素执行一个异步操作（如对 url 执行 http 请求）：flatMap+ 一个异步的返回类型为 Publisher的方法
  • …忽略一些数据：在 flatMap lambda 中根据条件返回一个 Mono.empty()
  • …保留原来的序列顺序：Flux#flatMapSequential（对每个元素的异步任务会立即执行，但会将结果按照原序列顺序排序）
  • …当 Mono 元素的异步任务会返回多个元素的序列时：Mono#flatMapMany
• 我想添加一些数据元素到一个现有的序列：
  • 在开头添加：Flux#startWith(T...)
  • 在最后添加：Flux#concatWith(T...)
• 我想将 Flux转化为集合（一下都是针对 Flux的）
  • 转化为 List：collectList，collectSortedList
  • 转化为 Map：collectMap，collectMultiMap
  • 转化为自定义集合：collect
  • 计数：count
  • reduce 算法（将上个元素的reduce结果与当前元素值作为输入执行reduce方法，如sum） reduce
  • …将每次 reduce 的结果立即发出：scan
  • 转化为一个 boolean 值：
  • 对所有元素判断都为true：all
  • 对至少一个元素判断为true：any
  • 判断序列是否有元素（不为空）：hasElements
  • 判断序列中是否有匹配的元素：hasElement
• 我想合并 publishers…
  • 按序连接：Flux#concat或 .concatWith(other)
  • …即使有错误，也会等所有的 publishers 连接完成：Flux#concatDelayError
  • …按订阅顺序连接（这里的合并仍然可以理解成序列的连接）：Flux#mergeSequential
  • 按元素发出的顺序合并（无论哪个序列的，元素先到先合并）：Flux#merge/ .mergeWith(other)
  • …元素类型会发生变化：Flux#zip/ Flux#zipWith
  • 将元素组合：
  • 2个 Monos 组成1个 Tuple2：Mono#zipWith
  • n个 Monos 的元素都发出来后组成一个 Tuple：Mono#zip
  • 在终止信号出现时“采取行动”：
  • 在 Mono 终止时转换为一个 Mono：Mono#and
  • 当 n 个 Mono 都终止时返回 Mono：Mono#when
  • 返回一个存放组合数据的类型，对于被合并的多个序列：
    • 每个序列都发出一个元素时：Flux#zip
    • 任何一个序列发出元素时：Flux#combineLatest
  • 只取各个序列的第一个元素：Flux#first，Mono#first，mono.or(otherMono).or(thirdMono)，flux.or(otherFlux).or(thirdFlux)
  • 由一个序列触发（类似于 flatMap，不过“喜新厌旧”）：switchMap
  • 由每个新序列开始时触发（也是“喜新厌旧”风格）：switchOnNext
• 我想重复一个序列：repeat
  • …但是以一定的间隔重复：Flux.interval(duration).flatMap(tick -> myExistingPublisher)
• 我有一个空序列，但是…
  • 我想要一个缺省值来代替：defaultIfEmpty
  • 我想要一个缺省的序列来代替：switchIfEmpty
• 我有一个序列，但是我对序列的元素值不感兴趣：ignoreElements
  • …并且我希望用 Mono来表示序列已经结束：then
  • …并且我想在序列结束后等待另一个任务完成：thenEmpty
  • …并且我想在序列结束之后返回一个 Mono：Mono#then(mono)
  • …并且我想在序列结束之后返回一个值：Mono#thenReturn(T)
  • …并且我想在序列结束之后返回一个 Flux：thenMany
• 我有一个 Mono 但我想延迟完成…
  • …当有1个或N个其他 publishers 都发出（或结束）时才完成：Mono#delayUntilOther
  • …使用一个函数式来定义如何获取“其他 publisher”：Mono#delayUntil(Function)
• 我想基于一个递归的生成序列的规则扩展每一个元素，然后合并为一个序列发出：
  • …广度优先：expand(Function)
  • …深度优先：expandDeep(Function)

3）“窥视”（只读）序列

• 再不对序列造成改变的情况下，我想：
  • 得到通知或执行一些操作：
  • 发出元素：doOnNext
  • 序列完成：Flux#doOnComplete，Mono#doOnSuccess
  • 因错误终止：doOnError
  • 取消：doOnCancel
  • 订阅时：doOnSubscribe
  • 请求时：doOnRequest
  • 完成或错误终止：doOnTerminate（Mono的方法可能包含有结果）
    • 但是在终止信号向下游传递 之后：doAfterTerminate
  • 所有类型的信号（Signal）：Flux#doOnEach
  • 所有结束的情况（完成complete、错误error、取消cancel）：doFinally
  • 记录日志：log
• 我想知道所有的事件:
  • 每一个事件都体现为一个 single对象：
  • 执行 callback：doOnEach
  • 每个元素转化为single对象：materialize
    • …在转化回元素：dematerialize
  • 转化为一行日志：log

4）过滤序列

• 我想过滤一个序列
  • 基于给定的判断条件：filter
  • …异步地进行判断：filterWhen
  • 仅限于指定类型的对象：ofType
  • 忽略所有元素：ignoreElements
  • 去重:
  • 对于整个序列：Flux#distinct
  • 去掉连续重复的元素：Flux#distinctUntilChanged
• 我只想要一部分序列：
  • 只要 N 个元素：
  • 从序列的第一个元素开始算：Flux#take(long)
    • …取一段时间内发出的元素：Flux#take(Duration)
    • …只取第一个元素放到 Mono中返回：Flux#next()
    • …使用 request(N)而不是取消：Flux#limitRequest(long)
  • 从序列的最后一个元素倒数：Flux#takeLast
  • 直到满足某个条件（包含）：Flux#takeUntil（基于判断条件），Flux#takeUntilOther（基于对 publisher 的比较）
  • 直到满足某个条件（不包含）：Flux#takeWhile
  • 最多只取 1 个元素：
  • 给定序号：Flux#elementAt
  • 最后一个：

.takeLast(1)

• …如果为序列空则发出错误信号：Flux#last()
• …如果序列为空则返回默认值：Flux#last(T)
• 跳过一些元素：
• 从序列的第一个元素开始跳过：Flux#skip(long)
  • …跳过一段时间内发出的元素：Flux#skip(Duration)
• 跳过最后的 n 个元素：Flux#skipLast
• 直到满足某个条件（包含）：Flux#skipUntil（基于判断条件），Flux#skipUntilOther（基于对 publisher 的比较）
• 直到满足某个条件（不包含）：Flux#skipWhile
• 采样：
• 给定采样周期：Flux#sample(Duration)
  • 取采样周期里的第一个元素而不是最后一个：sampleFirst
• 基于另一个 publisher：Flux#sample(Publisher)
• 基于 publisher“超时”：Flux#sampleTimeout（每一个元素会触发一个 publisher，如果这个 publisher 不被下一个元素触发的 publisher 覆盖就发出这个元素）
• 我只想要一个元素（如果多于一个就返回错误）…
  • 如果序列为空，发出错误信号：Flux#single()
  • 如果序列为空，发出一个缺省值：Flux#single(T)
  • 如果序列为空就返回一个空序列：Flux#singleOrEmpty

5）错误处理

• 我想创建一个错误序列：error…
  • …替换一个完成的 Flux：.concat(Flux.error(e))
  • …替换一个完成的 Mono：.then(Mono.error(e))
  • …如果元素超时未发出：timeout
  • …“懒”创建：error(Supplier)
• 我想要类似 try/catch 的表达方式：
  • 抛出异常：error
  • 捕获异常：
  • 然后返回缺省值：onErrorReturn
  • 然后返回一个 Flux或 Mono：onErrorResume
  • 包装异常后再抛出：.onErrorMap(t -> new RuntimeException(t))
  • finally 代码块：doFinally
  • Java 7 之后的 try-with-resources 写法：using工厂方法
• 我想从错误中恢复…
  • 返回一个缺省的：
  • 的值：onErrorReturn
  • Publisher：Flux#onErrorResume和 Mono#onErrorResume
  • 重试：retry
  • …由一个用于伴随 Flux 触发：retryWhen
• 我想处理回压错误（向上游发出“MAX”的 request，如果下游的 request 比较少，则应用策略）…
  • 抛出 IllegalStateException：Flux#onBackpressureError
  • 丢弃策略：Flux#onBackpressureDrop
  • …但是不丢弃最后一个元素：Flux#onBackpressureLatest
  • 缓存策略（有限或无限）：Flux#onBackpressureBuffer
  • …当有限的缓存空间用满则应用给定策略：Flux#onBackpressureBuffer带有策略 BufferOverflowStrategy

6） 基于时间的操作

• 我想将元素转换为带有时间信息的 Tuple2…
  • 从订阅时开始：elapsed
  • 记录时间戳：timestamp
• 如果元素间延迟过长则中止序列：timeout
• 以固定的周期发出元素：Flux#interval
• 在一个给定的延迟后发出 0：static Mono.delay.
• 我想引入延迟：
  • 对每一个元素：Mono#delayElement，Flux#delayElements
  • 延迟订阅：delaySubscription

7）拆分 Flux

• 我想将一个 Flux拆分为一个 Flux>：
  • 以个数为界：window(int)
  • …会出现重叠或丢弃的情况：window(int, int)
  • 以时间为界：window(Duration)
  • …会出现重叠或丢弃的情况：window(Duration, Duration)
  • 以个数或时间为界：windowTimeout(int, Duration)
  • 基于对元素的判断条件：windowUntil
  • …触发判断条件的元素会分到下一波（cutBefore变量）：.windowUntil(predicate, true)
  • …满足条件的元素在一波，直到不满足条件的元素发出开始下一波：windowWhile（不满足条件的元素会被丢弃）
  • 通过另一个 Publisher 的每一个 onNext 信号来拆分序列：window(Publisher)，windowWhen
• 我想将一个 Flux的元素拆分到集合…
  • 拆分为一个一个的 List:
  • 以个数为界：buffer(int)
    • …会出现重叠或丢弃的情况：buffer(int, int)
  • 以时间为界：buffer(Duration)
    • …会出现重叠或丢弃的情况：buffer(Duration, Duration)
  • 以个数或时间为界：bufferTimeout(int, Duration)
  • 基于对元素的判断条件：bufferUntil(Predicate)
    • …触发判断条件的元素会分到下一个buffer：.bufferUntil(predicate, true)
    • …满足条件的元素在一个buffer，直到不满足条件的元素发出开始下一buffer：bufferWhile(Predicate)
  • 通过另一个 Publisher 的每一个 onNext 信号来拆分序列：buffer(Publisher)，bufferWhen
  • 拆分到指定类型的 “collection”：buffer(int, Supplier)
• 我想将 Flux中具有共同特征的元素分组到子 Flux：groupBy(Function)（注意返回值是 Flux>，每一个 GroupedFlux具有相同的 key 值 K，可以通过 key()方法获取）。

8）回到同步的世界

• 我有一个 Flux，我想：
  • 在拿到第一个元素前阻塞：Flux#blockFirst
  • …并给出超时时限：Flux#blockFirst(Duration)
  • 在拿到最后一个元素前阻塞（如果序列为空则返回 null）：Flux#blockLast
  • …并给出超时时限：Flux#blockLast(Duration)
  • 同步地转换为 Iterable：Flux#toIterable
  • 同步地转换为 Java 8 Stream：Flux#toStream
• 我有一个 Mono，我想：
  • 在拿到元素前阻塞：Mono#block
  • …并给出超时时限：Mono#block(Duration)
  • 转换为 CompletableFuture：Mono#toFuture

处理错误

在响应式流中，错误（error）是终止（terminal）事件。

当有错误发生时，它会导致流序列停止， 并且错误信号会沿着操作链条向下传递，直至遇到定义的 Subscriber 及其 onError 方法。

在 try-catch 代码块中处理异常的几种方法。常见的包括如下几种：

1. 捕获并返回一个静态的缺省值。
2. 捕获并执行一个异常处理方法。
3. 捕获并动态计算一个候补值来顶替。
4. 捕获，并再包装为某一个 业务相关的异常，然后再抛出业务异常。
5. 捕获，记录错误日志，然后继续抛出。
6. 使用 finally 来清理资源

以上所有这些在 Reactor 都有等效的 操作符处理方式。

与第 (1) 条（捕获并返回一个静态的缺省值）对应的是 onErrorReturn：

Flux.just(10)   .flatMap(this::function)   .onErrorReturn("Error");//返回一个静态的缺省值

根据错误类型返回对应值

Flux.just(10)   .flatMap(this::function)   .onErrorReturn(e -> e.getMessage().equals("boom-1"), "Error-1");

与第 (2、3、4) 条（捕获并执行一个异常处理方法）对应的是 onErrorResume

Flux.just(10)   .flatMap(m -> function(k).onErrorResume(e -> handleErr(k)));Flux.just(10)   .flatMap(m -> function(k).onErrorResume(e -> errFunc(k)));Flux.just(10)   .flatMap(m -> function(k)     .onErrorResume(e -> Flux.error(new Exception(k)))   );

对应第 (5) 条（捕获，记录错误日志，并继续抛出）

Flux.just(10)   .flatMap(k -> function(k))    .doOnError(e ->     log.error(e.getLocalizedMessage(), e);   });

对应（6）doFinally 在序列终止（无论是 onComplete、onError还是取消）的时候被执行， 并且能够判断是什么类型的终止事件

Flux.just(10)    .doFinally(type -> {        if (type == SignalType.CANCEL){            log.info("a log");         }    }) 

流的并行（或者异步）与多线程

Reactor被视为与并发无关的。

也就是说，获得Flux或Mono并不一定意味着它在专用线程中运行。

取而代之的是，大多数Operator会继续在执行前一个Operator的线程中工作。

除非指定，否则最顶层的Operator（源）本身运行在进行subscribe()调用的线程上。

先分享一个案例：

在进行DB这种耗时的操作时，我们不希望在IO线程上执行，而是在专用的线程池里边执行，从而使得IO线程不会阻塞,。

来看看下面的代码：

//获得所有的用户public Mono getAllUser(ServerRequest request) {    return ServerResponse.ok()        .contentType(MediaType.APPLICATION_JSON)        .body(userRepository.findAll(), User.class);}

代码中没有使用subscriptOn或者publishOn操作符，userRepository.findAll() 会使后续的操作符都在 lettuce-1 线程（IO）线程中执行，所以， 并且整个流的执行都是串行的，

我们需要将查询过程改为异步查询。

调度程序(Schedulers)和线程(threads)

Project Reactor中的调度程序(Schedulers)用于指示多线程(multi-threading)。

某些运算符具有将Scheduler作为参数的变体。 这些指示操作员在特定的调度程序上执行其部分或全部工作。

Project Reactor附带了一组预配置的调度程序(Schedulers)，都可以通过Schedulers类进行访问：

• Schedulers.parallel(): 执行诸如事件循环和回调处理之类的计算工作。
• Schedulers.immediate(): 在当前线程中立即执行工作
• Schedulers.elastic(): 执行I/O绑定的工作，例如阻塞I/O的异步性能，此调度程序由线程池支持，该线程池将根据需要增长
• Schedulers.newSingle(): 在新线程上执行工作
• Schedulers.fromExecutor(): 从java.util.concurrent.Executor创建调度程序
• Schedulers.timer(): 创建或重新使用分辨率为50ms的基于hash-wheel的TimedScheduler。

不要将计算调度程序用于I/O。

调度程序可以通过以下几种不同的方式执行发布者：

• 使用利用调度程序的运算符
• 明确地通过将调度程序传递给这样的运算符
• 通过使用 subscribeOn(Scheduler)
• 通过使用 publishOn(Scheduler)

如果没有其他说明，默认情况下，诸如buffer, replay, skip, delay, parallel等操作符将使用调度程序。

如果需要，所有列出的运算符都允许你传入自定义调度程序。 大多数时候都使用默认值是一个好主意。

如果希望在特定的调度程序上执行订阅链，请使用subscribeOn()运算符。

该代码在未设置调度程序的情况下在主线程上执行：

Flux.just("Ben", "Michael", "Mark").flatMap(key -> {            System.out.println("Map 1: " + key + " (" + Thread.currentThread().getName() + ")");            return Flux.just(key);        }).flatMap(value -> {            System.out.println("Map 2: " + value + " (" + Thread.currentThread().getName() + ")");            return Flux.just(value);        }).subscribe();

该示例打印以下行：

Map 1: Ben (main)Map 2: Ben (main)Map 1: Michael (main)Map 2: Michael (main)Map 1: Mark (main)Map 2: Mark (main)

此示例显示了添加到流中的subscribeOn()方法（在哪里添加都无所谓）：

Flux.just("Ben", "Michael", "Mark").flatMap(key -> {            System.out.println("Map 1: " + key + " (" + Thread.currentThread().getName() + ")");            return Flux.just(key);        }).flatMap(value -> {            System.out.println("Map 2: " + value + " (" + Thread.currentThread().getName() + ")");            return Flux.just(value);        }).subscribeOn(Schedulers.parallel()).subscribe();

该示例的输出显示了subscribeOn()的效果。

你可以看到Publisher在同一线程上执行，但在计算线程池上执行：

Map 1: Ben (parallel-1)Map 2: Ben (parallel-1)Map 1: Michael (parallel-1)Map 2: Michael (parallel-1)Map 1: Mark (parallel-1)Map 2: Mark (parallel-1)

如果将相同的代码应用于Lettuce，你将注意到执行第二个flatMap()的线程有所不同：

Flux.just("Ben", "Michael", "Mark").flatMap(key -> {    System.out.println("Map 1: " + key + " (" + Thread.currentThread().getName() + ")");    return commands.set(key, key);}).flatMap(value -> {    System.out.println("Map 2: " + value + " (" + Thread.currentThread().getName() + ")");    return Flux.just(value);}).subscribeOn(Schedulers.parallel()).subscribe();

该示例打印以下行：

Map 1: Ben (parallel-1)Map 1: Michael (parallel-1)Map 1: Mark (parallel-1)Map 2: OK (lettuce-nioEventLoop-3-1)Map 2: OK (lettuce-nioEventLoop-3-1)Map 2: OK (lettuce-nioEventLoop-3-1)

与独立示例有两点不同：

• 这些值是同时设置的，而不是顺序设置的
• 第二个flatMap()转换输出netty EventLoop线程名称这是因为默认情况下，Lettuce发布者是在netty EventLoop线程上执行和完成的。

publishOn指示发布者在特定的Scheduler上调用其观察者的nNext, onError和onCompleted方法。

在这里，顺序很重要：

Flux.just("Ben", "Michael", "Mark").flatMap(key -> {    System.out.println("Map 1: " + key + " (" + Thread.currentThread().getName() + ")");    return commands.set(key, key);}).publishOn(Schedulers.parallel()).flatMap(value -> {    System.out.println("Map 2: " + value + " (" + Thread.currentThread().getName() + ")");    return Flux.just(value);}).subscribe();

publishOn()调用之前的所有操作均在main中执行，而调度器中的以下所有操作均在其中执行：

Map 1: Ben (main)Map 1: Michael (main)Map 1: Mark (main)Map 2: OK (parallel-1)Map 2: OK (parallel-1)Map 2: OK (parallel-1)

调度程序(Schedulers)允许直接调度操作。 有关更多信息，请参考Project Reactor文档。

两种在流中切换执行上下文（或Scheduler）的方式

Reactor提供了两种在流中切换执行上下文（或Scheduler）的方式：

• publishOn
• subscribeOn

我们可以通过这两种方式达到异步执行的目的

publishOn :

此运算符影响线程上下文，它下面的链中的其余运算符将在其中执行，直到新出现的publishOn。

void demo() {    final Flux flux = Flux.interval(Duration.of(100, ChronoUnit.MILLIS))            .map(log())//1            .publishOn(Schedulers.parallel())//2            .map(log())//3            .publishOn(Schedulers.elastic())//4            .flatMap(logOfMono());//5    flux.subscribe();}

上面代码中：

1. 会在调用flux.subscribe()的线程上执行
2. 指定后续的运算符执行的上下文
3. 会在parallel线程上执行
4. 指定后续的运算符执行的上下文
5. 会在elastic线程上执行

subscribeOn ：

整个流在指定的Scheduler的Scheduler.Worker上运行 ，直至出现publishOn ，publishOn后续的操作符由 publishOn 决定执行上下文。

void demo() {    final Flux flux = Flux.interval(Duration.of(100, ChronoUnit.MILLIS))            .map(log())//1            .publishOn(Schedulers.parallel())            .map(log())//2            .subscribeOn(Schedulers.elastic())            .flatMap(logOfMono());    flux.subscribe();}

上面代码中：

1. 会在elastic线程执行
2. 会在parallel线程上执行，publishOn会“覆盖”subscribeOn的行为

这样我们就可以异步的去执行方法了。

Flux依次发出了1，2，3三个元素 无论是subscribeOn、还是publishOn，经过flatMap 都会在 parallel-1 线程上执行，也就是说，Flux中的所有元素只是从主线程发出，在另一个线程中执行。

如何实现流的并行

在调用一些阻塞方法时（rpc、redis、io），我们期望每个元素经过 flatMap 中时可以运行在不同线程上（串行-> 并发），应该怎么做？

并行的方法

• 1 使用ParallelFlux

  将Flux转为ParallelFlux，使用runOn来指明需要的线程池

• 2 内部通过publishOn表明执行上下文

fluxMap中调用的方法需要在内部通过publishOn表明执行上下文。

• 3 创建异步流

1使用ParallelFlux

将Flux转为ParallelFlux，使用runOn来指明需要的线程池

要获得ParallelFlux，可以在Flux 上使用parallel()运算符。

为了告诉ParallelFlux在哪里运行每个元素，必须使用runOn(Scheduler)。

如果并行处理后，您想恢复到“正常”状态 Flux并以顺序方式应用运算符链的其余部分，可以使用sequential()

void demo() {    final Flux flux = Flux.fromIterable(Lists.newArrayList(3L, 1L, 2L))            .parallel().runOn(Schedulers.elastic())            .flatMap(logOfMono())            .sequential();    flux.subscribe();}

2内部通过publishOn表明执行上下文

fluxMap中调用的方法需要在内部通过publishOn表明执行上下文。

void demo() {    Flux.fromIterable(Lists.newArrayList(3L, 1L, 2L))        .flatMap(this::blockFunction)        .subscribe();}//通过publishOn 来表明 blockFunction() 的上下文Mono blockFunctionAsync(Long i) {    return Mono.just(i).publishOn(Schedulers.elastic()).flatMap(this::blockFunction);}Mono blockFunction(Long i) {    Thread.sleep(i * 1000);    return Mono.just(i);}

3创建异步流

1. 使用 Mono.fromFuture() 创建流

Mono.fromFuture(CompletableFuture.supplyAsync(() -> blockFunction()));

Mono.fromFuture() 创建一个Mono ，需要提供一个 CompletableFuture 产生它的值。

CompletableFuture.supplyAsync() 返回一个 CompletableFuture，它将在ForkJoinPool.commonPool() 上运行任务，异步完成。

ForkJoinPool一个全局线程池，主要应用于计算密集型的场景。

3.2.使用Mono.create() 、Flux.create() 方法创建流

//自定义线程池final Executor executor = new ThreadPoolExecutor(...); //创建Mono、Flux时 指明上下文Mono.create(sink -> executor.execute(() -> {    sink.success(blockFunction());}));            Mono userMono = Mono.create(sink -> bizPool.execute(() -> {    sink.success( jpaEntityService.selectOne(dto.getUserId()));})); 

目前我们推荐使用 这种方法来创建Mono、Flux来实现异步

这样做的好处：

1. Reactor对新手来说有一定理解成本，在调用一个返回值为 Publisher 的类型的方法的时候，不点进去看看，无法知道这是同步方法还是异步响应式方法，这样显示开发人员知道这是一个 响应式方法，避免对于调用者造成的心智负担。
2. 更方便得使用自定义线程池

以上3中流并行方法，如何选择？

建议查询业务场景 采取合适的方法：

方法1 ：

ParallelFlux 不能保证上游元素 返回顺序，不满足多个 sql 有序执行，或者 上游 元素有序执行的 场景。

方法2：

Reactor对新手来说有一定理解成本，在调用一个返回值为 Publisher 的类型的方法时，不点进去无法知道这是同步方法还是异步方法，对于调用者存在心智负担。

方法3.1：

无法指定自定义线程池

方法3.2：

这样做的好处：

1. Reactor对新手来说有一定理解成本，在调用一个返回值为 Publisher 的类型的方法的时候，不点进去看看，无法知道这是同步方法还是异步响应式方法，这样显示开发人员知道这是一个 响应式方法，避免对于调用者造成的心智负担。
2. 更方便得使用自定义线程池

Webflux 概述

简单来说，Webflux 是响应式编程的框架，与其对等的概念是 SpringMVC。

两者的不同之处在于 Webflux 框架是异步非阻塞的，其可以通过较少的线程处理高并发请求。

Webflux 的框架底层采用了 Reactor响应式编程框架以及 Netty，关于这两部分内容可以参看我之前的3高基础书籍：

《Java高并发核心编程 卷1 加强版》

作为一个异步框架来说，必须保证整个程序链中的每一步都是异步操作，如果在某一步出现了同步阻塞（如等待数据库 IO），则整个程序还是回出现阻塞的问题。

webflux应用场景

既适合IO密集型、磁盘IO密集、网络IO密集等服务场景，也适用于高并发、高吞吐业务场景的响应式改造。

比如微服务网关，就可以使用webflux技术来显著的提升网关对下游服务的吞吐量，spring cloud gateway就使用了webflux这门技术

Spring boot webflux介绍

先提一下Spring Boot 2.0

spring.io 官网有句醒目的话是：

BUILD ANYTHING WITH SPRING BOOT

Spring Boot （Boot 顾名思义，是引导的意思）框架是用于简化 Spring 应用从搭建到开发的过程。

应用开箱即用，只要通过一个指令，包括命令行 java -jar 、SpringApplication 应用启动类 、 Spring Boot Maven 插件等，就可以启动应用了。

另外，Spring Boot 强调只需要很少的配置文件，所以在开发生产级 Spring 应用中，让开发变得更加高效和简易。

目前，Spring Boot 2.x 包括 WebFlux。

Spring Boot 2.0 WebFlux

Spring Boot Webflux 就是基于 Reactive Streams 实现的。Spring Boot 2.0 包括一个新的 spring-webflux 模块。

所以，了解 WebFlux，首先了解下什么是 Reactive Streams。

Reactive Streams 是 JVM 中面向流的库标准和规范：

• 处理可能无限数量的元素
• 按顺序处理
• 组件之间异步传递
• 强制性非阻塞背压（Backpressure）

Backpressure（背压）

背压是一种常用策略，使得发布者拥有无限制的缓冲区存储元素，用于确保发布者发布元素太快时，不会去压制订阅者。

响应式 API

Reactor 框架是Reactive Streams 的一个实现框架，也是 Spring Boot Webflux 响应库依赖，通过 Reactor 并与其他响应库交互。

Webflux 提供了 两种响应式 API：Mono 和 Flux。

Webflux 的处理流程是：一般是将 Publisher 作为输入，在框架内部转换成 Reactor 类型并处理逻辑，然后返回 Flux 或 Mono 作为输出。

Spring Webflux

Spring Boot Webflux 有两种编程模型实现，一种类似 Spring MVC 注解方式，另一种是使用其功能性端点方式。

Spring Boot 2.0 WebFlux 特性

常用的 Spring Boot 2.0 WebFlux 生产的特性如下：

• 响应式 API
• 编程模型
• 适用性
• 内嵌容器
• Starter 组件

还有对日志、Web、消息、测试及扩展等支持。

spring webflux和spring mvc的异同点

WebFlux 和 MVC 的交集

一图就很明确了，WebFlux 和 MVC 有交集，方便大家迁移。但是注意：

• MVC 能满足场景的，就不需要更改为 WebFlux。
• 要注意容器的支持，可以看看下面内嵌容器的支持。
• 微服务体系结构，WebFlux 和 MVC 可以混合使用。尤其开发 IO 密集型服务的时候，选择 WebFlux 去实现。
• spring mvc是一个命令式的编程方式采用同步阻塞方式，方便开发人员编写代码和调试；spring webflux调试会非常不方便
• JDBC连接池和JPA等技术还是阻塞模型，传统的关系型数据库如MySQL也不支持非阻塞的方式获取数据，目前只有非关系型数据库如Redis、Mongodb支持非阻塞方式获取数据

编程模型

Spring 5 web 模块包含了 Spring WebFlux 的 HTTP 抽象。类似 Servlet API , WebFlux 提供了 WebHandler API 去定义非阻塞 API 抽象接口。可以选择以下两种编程模型实现：

• 注解控制层。和 MVC 保持一致，WebFlux 也支持响应性 @RequestBody 注解。
• 功能性端点。基于 lambda 轻量级编程模型，用来路由和处理请求的小工具。和上面最大的区别就是，这种模型，全程控制了请求 - 响应的生命流程

内嵌容器

跟 Spring Boot 大框架一样启动应用，但 WebFlux 默认是通过 Netty 启动，并且自动设置了默认端口为 8080。另外还提供了对 Jetty、Undertow 等容器的支持。

开发者自行在添加对应的容器 Starter 组件依赖，即可配置并使用对应内嵌容器实例。

但是要注意，必须是 Servlet 3.1+ 容器，如 Tomcat、Jetty；或者非 Servlet 容器，如 Netty 和 Undertow。

Netty优点

• API使用简单、易上手
• 功能强大、支持多种主流协议
• 定制能力强、可扩展性高
• 性能高、综合性能最优
• 成熟稳定、久经考验
• 社区活跃、学习资料多

Netty selector模型

SpringMVC 与Spring WebFlux 的根本区别

Spring Webflux模块包含对响应式 HTTP 和 WebSocket 客户端的支持，以及对 REST，HTML 和 WebSocket 交互等程序的支持。

一般来说，Spring MVC 用于同步处理，Spring Webflux 用于异步处理。

回顾一下，传统的以SpringMVC为代表的webmvc技术使用的是同步阻塞式IO模型

而Spring WebFlux是一个异步非阻塞式IO模型，可以用少量的容器线程支撑大量的并发访问，

所以Spring WebFlux可以提升吞吐量和伸缩性，但是接口的响应时间并不会缩短，

其处理结果还是得由worker线程处理完成之后在返回给请求

Webflux 基本使用

首先创建 maven 项目，在项目的 pom 文件中引入相应的依赖

    org.springframework.boot    spring-boot-starter-parent    2.7.3                 org.springframework.boot        spring-boot-starter-data-r2dbc                org.springframework.boot        spring-boot-starter-webflux                mysql        mysql-connector-java        runtime                dev.miku        r2dbc-mysql        0.8.2.RELEASE                org.projectlombok        lombok        1.18.24    

创建项目的启动类

@SpringBootApplicationpublic class WebfluxDemoApplication {    public static void main(String[] args) {        SpringApplication.run(WebfluxDemoApplication.class);    }}

此时，我们就可以编写一个简单的 Controller 来感受一下 Webflux 框架异步相应的概念

package com.crazymaker.springcloud.reactive.user.info.controller;import com.crazymaker.springcloud.common.result.RestOut;import com.crazymaker.springcloud.common.util.ThreadUtil;import com.crazymaker.springcloud.reactive.user.info.dto.User;import com.crazymaker.springcloud.reactive.user.info.service.impl.JpaEntityServiceImpl;import io.swagger.annotations.Api;import io.swagger.annotations.ApiImplicitParam;import io.swagger.annotations.ApiImplicitParams;import io.swagger.annotations.ApiOperation;import lombok.extern.slf4j.Slf4j;import org.springframework.web.bind.annotation.*;import reactor.core.publisher.Flux;import reactor.core.publisher.Mono;import javax.annotation.Resource;import java.util.concurrent.ExecutorService;import java.util.function.Supplier;import static com.crazymaker.springcloud.reactive.user.info.parallel.DBFunctionWrapper.dbFunWrapSafe;/** * Mono 和 Flux 适用于两个场景，即： * Mono：实现发布者，并返回 0 或 1 个元素，即单对象。 * Flux：实现发布者，并返回 N 个元素，即 List 列表对象。 * 有人会问，这为啥不直接返回对象，比如返回 City/Long/List。 * 原因是，直接使用 Flux 和 Mono 是非阻塞写法，相当于回调方式。 * 利用函数式可以减少了回调，因此会看不到相关接口。这恰恰是 WebFlux 的好处：集合了非阻塞 + 异步 */@Slf4j@Api(value = "用户信息、基础学习DEMO", tags = {"用户信息DEMO"})@RestController@RequestMapping("/api/user")public class UserReactiveController {    static ExecutorService bizPool = ThreadUtil.getIoIntenseTargetThreadPool();    @ApiOperation(value = "回显测试", notes = "提示接口使用者注意事项", httpMethod = "GET")    @RequestMapping(value = "/hello")    @ApiImplicitParams({            @ApiImplicitParam(paramType = "query", dataType = "string", dataTypeClass = String.class, name = "name", value = "名称", required = true)})    public Mono> hello(@RequestParam(name = "name") String name) {        log.info("方法 hello 被调用了");        return Mono.just(RestOut.succeed("hello " + name));    }}

上述代码中，我们定义了一个异步响应的接口，启动程序调用相应接口，可以看到效果

WebFlux和Spring Web MVC 的组件对比

Spring WebFlux不依赖于Servlet API，它可以运行在非Servlet容器Netty、Undertow和任何Servlet 3.1+的Servlet容器（Tomcat，Jetty）之上。

虽然Servlet 3.1提供了非阻碍I/O的API，但是有很多其它的API依然是同步或阻碍式的；这使得Spring需要重新构建完全基于异步和非阻碍式的运行环境。

Spring WebFlux最底层的组件是HttpHandler，它用来适配不同的服务器引擎（Netty、Undertow、Tomcat、Jetty）。

在通过HttpHandler消除了服务器引擎的异构行后，Spring WebFlux的API设计与Spring MVC是高度一致的。

Spring WebFlux与Spring MVC在概念上有对应的关系：

Spring WebFlux支持两种编程模型：

• 注解控制器：和Spring MVC使用的注解式保持一致。

  • RequestMappingHandlerMapping：映射请求与@RequestMapping控制器类和方法；
  • RequestMappingHandlerAdapter：调用@RequestMapping注解的方法。

  Sping MVC的两个类与这两个类名称一致，在不同的包里，有不同的实现，但功能保持一致。

• 函数式端点：基于Lambada表达式、轻量级的函数式编程模型。

  • RouterFunctionMapping：用来支持RouterFunction；
  • HandlerFunctionAdapter：用来支持HandlerFunctions。

  从Spring 5.2或Spring 2.2.x以后Spring MVC也支持这种编程模型。

Spring Boot的自动配置

Spring Boot提供的自动配置主要有：

• CodecsAutoConfiguration：Spring WebFlux使用HttpMessageReader和HttpMessageWriter接口来转换HTTP请求和返回。本配置类为我们注册了CodecCustomizer的Bean，默认使用Jackson2JsonEncoder和Jackson2JsonDecoder。

• ReactiveWebServerFactoryAutoConfiguration：为响应式Web服务器进行自动配置。

• WebFluxAutoConfiguration：使用等同于@EnableWebFlux的配置开启WebFlux的支持。可通过WebFluxProperties使用spring.webflux.*来对WebFlux进行配置：

spring:    webflux:      date-format: yyyy-MM-dd # 日期格式      static-path-pattern: /resouces/static/** # 静态资源目录

• WebClientAutoConfiguration：为WebClient进行自动配置。

传统代码转为Reactive的桥梁

Mono/FluxCreate ——传统代码转为Reactive的桥梁

Reactive Programming 作为观察者模式（Observer） 的延伸，不同于传统的命令编程方式（ Imperative programming）同步拉取数据的方式，如迭代器模式（Iterator） 。

而是采用数据发布者同步或异步地推送到数据流（Data Streams）的方案。当该数据流（Data Steams）订阅者监听到传播变化时，立即作出响应动作。

在实现层面上，Reactive Programming 可结合函数式编程简化面向对象语言语法的臃肿性，屏蔽并发实现的复杂细节，提供数据流的有序操作，从而达到提升代码的可读性，以及减少 Bugs 出现的目的。

同时，Reactive Programming 结合背压（Backpressure）的技术解决发布端生成数据的速率高于订阅端消费的问题。

Project Reactor提供了很多创建Mono/Flux的静态方法，而最常用的就是Mono#create方法，通过该方法能把以前命令式的程序转化为Reactive的编程方式。

众所周知，Reactive Programming是一种Pull-Push模型，其中Pull用于实现back-pressure，

Iterator 属于推模式（push-based），Reactive Flux/Mono 属于拉模式（pull-based）

下面以一个常见的Pull模型迭代器Iterator来说明如何将传统代码转为Reactive的代码。

Iterator 推模式-> Flux 拉模式

//创建一个迭代器Iterator it = Arrays.asList<>(1,2,3).iterator();//使用迭代器while(it.hasNext()) {    //模拟业务逻辑 —— 这里直接打印value    System.out.println(it.next());}

上面是一个常见的迭代器使用方式，下面看看是如何将迭代器转换成Flux的:

@Testpublic void fluxExample() throws InterruptedException {    //创建迭代器    Iterator it = Arrays.asList(1, 2, 3).iterator();    Flux iteratorFlux = Flux.create(sink -> {        while (it.hasNext()) {            Integer data = (Integer) it.next();            sink.next(data); //利用FluxSink实现data的Push        }        sink.complete();  //发送结束的Signal    });    //进行订阅，进行业务逻辑操作    iteratorFlux.log().subscribe(System.out::println);}

输出

23:29:15.727 [main] INFO reactor.Flux.Create.1 - onSubscribe(FluxCreate.BufferAsyncSink)23:29:15.730 [main] INFO reactor.Flux.Create.1 - request(unbounded)23:29:20.931 [main] INFO reactor.Flux.Create.1 - onNext(1)123:29:20.931 [main] INFO reactor.Flux.Create.1 - onNext(2)223:29:20.931 [main] INFO reactor.Flux.Create.1 - onNext(3)323:29:20.932 [main] INFO reactor.Flux.Create.1 - onComplete()Disconnected from the target VM, address: "127.0.0.1:58005", transport: "socket"Process finished with exit code 0

MonoCreate常见的两者使用方式

传统命令式编程除了Iterator的Pull模式外，通常还有Observable以及Callback这两种Push模式，下面分别举例讲讲这两种模式。

Observable -> MonoCreate

Observable原始代码举例：

Observable observable = new Observable() {    //需要重写Observable，默认是setChanged与notifyObservers分离，实现先提交再通知的效果    //这里为了简单起见，将通知与提交放在了一起    @Override    public void notifyObservers(Object arg) {    setChanged();    super.notifyObservers(arg);    }};Observer first = (ob,value) -> {    System.out.println("value is " + value);};observable.addObserver(first);observable.notifyObservers("42");//    after some time, cancel observer to dispose resourceobservable.deleteObserver(first);

MonoCreate的转化示例：

Mono