springWebFlux介绍


是spring5添加的新模块,用于web开发,功能与springMvc类似,WebFlux是应付当前一种比较流行的响应式编程而出现的框架


使用传统web框架,比如SpringMVC,这些基于servlet容器,而WebFlux是一种异步非阻塞的框架


WebFlux特点

非阻塞式,在有限资源下,提供系统吞吐量和伸缩性,以Reactor为基础实现响应式编程
函数式编程,spring5框架基于java,WebFlux可以使用java8函数式编程方式实现路由请求



比较springMvc


[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-HQKdOaGj-1608174188913)(./webflux比较springmvc.jpg)]


都可以使用注解方式,都可以运行在tomcat等容器中


springmvc采用命令式编程,webflux采用异步响应式编程


使用场景,网关




什么是响应式编程

RP(Reactive Programming),响应式编程是一种面向数据流和变化传播的编程范式,这意味着可以在编程语言中很方便地表达静态或动态的数据流,而相关的计算模型会自动将变化的值通过数据流进行传播
java8及其之前版本

提供观察者模式的两个类Observer和Observable


java9 使用Flow接口取代Observer和Observable,Flow才是真正意义上的响应式编程



响应式编程(Reactor实现)

响应式编程操作中,Reactor是盲仔Reactive规范的框架
Reactor有两个核心的类,Mono和Flux,这个两个类都实现了Publisher接口,提供了丰富操作符,Flux对象实现发布者,返回N个元素;Mono实现发布者,返回0或1一个元素
Flux和Mono都是数据流的发布者,使用Flux和Mono都可以发出三种数据信号,元素值,错误信号,完成信号,错误信号和完成信号都代表终止信号,终止信号用于高速订阅者数据流结束了,错误信号在终止数据流的同时会把错误信息传递给订阅者
三种信号的特点

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

1、Webflux



用于web开发，功能和springMVC类似，响应式编程实现，异步非阻塞。

关于异步非阻塞：

*异步和同步：都是针对调用者。调用者发送请求，如果等到对方回应之后才去做其他事情就是同步。如果发送请求后不用等就去做其他事，就是异步。

*阻塞和非阻塞：都是针对被调用者，被调用者收到请求之后，做完事情之后才给出反馈就是阻塞。被调用者收到请求之后，马上给出反馈再做事情就是非阻塞。

Webflux优势：

*异步非阻塞，有限资源下提高系统吞吐量和伸缩性，以Reactor为基础实现响应式编程

*函数式编程：Spring5基于java8，所以webflux可以用java8的函数式编程方式实现路由请求。

Spring webFlux和SpringMVC比较：两个框架都可以使用注解，都运行在Tomcat容器。但SpringMVC是命令式编程，webflux是异步响应式编程。

 



 

2、响应式编程：

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

（1）Java8及以前版本：

提供了两个观察者模式两个类：Observer和Observable

public class ObserverDemo extends Observable {
    public static void main(String[] args) {
        ObserverDemo observer=new ObserverDemo();
        // add observer
        observer.addObserver((o, arg) -> {
            System.out.println("发生变化");
        });
        observer.addObserver((o, arg) -> {
            System.out.println("收到被观察者通知，准备改变");
        });
        observer.setChanged();// data changing
        observer.notifyObservers();//notify
    }
}

（2）Flow类取代了Java9。两个方法：Subscribe订阅和Publish发布

 

3、Webflux执行流程和核心API

4、SpringWebflux（基于注解实现）

5、SpringWebflux（基于函数式编程实现）

webflux


官方文档

https://projectreactor.io/docs/core/release/api/reactor/core/publisher/Flux.html#zipWith-org.reactivestreams.Publisher-


写的比较好的文章

https://cloud.tencent.com/developer/article/1526029


介绍
Reactor中的发布者（Publisher）由Flux和Mono两个类定义，它们都提供了丰富的操作符（operator）。一个Flux对象代表一个包含0…N个元素的响应式序列，而一个Mono对象代表一个包含零/一个（0…1）元素的结果。
一、 创建数据流
Flux.just(1, 2, 3, 4, 5, 6);
Mono.just(1);

Flux和Mono提供了多种创建数据流的方法，just就是一种比较直接的声明数据流的方式，其参数就是数据元素。
对于图中的Flux，还可以通过如下方式声明（数组、集合、Stream、range）：
Integer[] array = new Integer[]{1,2,3,4,5,6};
Flux.fromArray(array);
List<Integer> list = Arrays.asList(array);
Flux.fromIterable(list);
Stream<Integer> stream = list.stream();
Flux.fromStream(stream);
Flux<Integer> rangeFlux =Flux.range(0, 10000);


二、 完成信号和错误信号
既然是“数据流”的发布者，Flux和Mono都可以发出三种“数据信号”：元素值、错误信号、完成信号，错误信号和完成信号都是终止信号，完成信号用于告知下游订阅者该数据流正常结束，错误信号终止数据流的同时将错误传递给下游订阅者。
不过，这三种信号都不是一定要具备的：
首先，错误信号和完成信号都是终止信号，二者不可能同时共存；

如果没有发出任何一个元素值，而是直接发出完成/错误信号，表示这是一个空数据流；

如果没有错误信号和完成信号，那么就是一个无限数据流。

比如，对于只有完成/错误信号的数据流：
// 只有完成信号的空数据流
Flux.just();
Flux.empty();
Mono.empty();
Mono.justOrEmpty(Optional.empty());
// 只有错误信号的数据流
Flux.error(new Exception("some error"));
Mono.error(new Exception("some error"));

三、订阅前什么都不会发生
数据流有了，假设我们想把每个数据元素原封不动地打印出来：
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()方法调用的时候才会触发数据流。所以，订阅前什么都不会发生。
webflux中的subscribe
在webFlux框架中 我们都是返回mono或flux类型的result， 不需要主动调用subscribe方法，
四、操作符
1. map - 元素映射为新元素
map操作可以将数据元素进行转换/映射，得到一个新元素。
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的函数式接口为参数，这个函数式的作用是定义转换操作的策略。举例说明：
StepVerifier.create(Flux.range(1, 6)    
            .map(i -> i * i))   
            .expectNext(1, 4, 9, 16, 25, 36)    
            .expectComplete();  

Flux.range(1, 6)用于生成从“1”开始的，自增为1的“6”个整型数据；

map接受lambda i -> i * i为参数，表示对每个数据进行平方；

验证新的序列的数据；

verifyComplete()相当于expectComplete().verify()。
2、flatMap - 元素映射为流
flatMap操作可以将每个数据元素转换/映射为一个流，然后将这些流合并为一个大的数据流。
flatMap
注意到，流的合并是异步的，先来先到，并非是严格按照原始序列的顺序（如图蓝色和红色方块是交叉的）。
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();

对于每一个字符串s，将其拆分为包含一个字符的字符串流；

对每个元素延迟100ms；

对每个元素进行打印（ps: doOnNext方法是“偷窥式”的方法，不会消费数据流）；

验证是否发出了8个元素。

打印结果为mfolnuox，原因在于各个拆分后的小字符串都是间隔100ms发出的，因此会交叉。
flatMap通常用于每个元素又会引入数据流的情况，比如我们有一串url数据流，需要请求每个url并收集respon<se数据。假设响应式的请求方法如下：
Mono requestUrl(String url) {…}

而url数据流为一个Flux urlFlux，那么为了得到所有的HttpResponse，就需要用到flatMap：
urlFlux.flatMap(url -> requestUrl(url));

其返回内容为Flux类型的HttpResponse流。
3）filter - 过滤
filter操作可以对数据元素进行筛选。

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();

filter的lambda参数表示过滤操作将保留奇数；

验证仅得到奇数的平方。
4）zip - 一对一合并
看到zip这个词可能会联想到拉链，它能够将多个流一对一的合并起来。zip有多个方法变体，我们介绍一个最常见的二合一的。
ps:注意对多个流进行zip后，合并后的长度跟这些流中最短的那个长度是一致的。有可能会丢数据，这点要特别注意
zip它对两个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
            .subscribe(t -> System.out.println(t.getT1()), null, countDownLatch::countDown);    // 4
    countDownLatch.await(10, TimeUnit.SECONDS);     // 5
}

将英文说明用空格拆分为字符串流；

定义一个CountDownLatch，初始为1，则会等待执行1次countDown方法后结束，不使用它的话，测试方法所在的线程会直接返回而不会等待数据流发出完毕；

使用Flux.interval声明一个每200ms发出一个元素的long数据流；因为zip操作是一对一的，故而将其与字符串流zip之后，字符串流也将具有同样的速度；

zip之后的流中元素类型为Tuple2，使用getT1方法拿到字符串流的元素；定义完成信号的处理为countDown;

countDownLatch.await(10, TimeUnit.SECONDS)会等待countDown倒数至0，最多等待10秒钟。

除了zip静态方法之外，还有zipWith等非静态方法，效果与之类似：
getZipDescFlux().zipWith(Flux.interval(Duration.ofMillis(200)))

在异步条件下，数据流的流速不同，使用zip能够一对一地将两个或多个数据流的元素对齐发出。
5）错误处理
onErrorReturn 返回静态值
onErrorResume 可以动态的返回
public static final Function<Throwable, Mono<ServerResponse>> onErrorResume = throwable -> {
        log.warn("", throwable);
        return ServerResponse.ok().contentType(MediaType.APPLICATION_JSON).body(
                fromObject(Result.of(-1, "异常，", throwable.getMessage())));
    };

onErrorReturn 和 onErrorResume 有什么区别吗？
从源码上看它们的区别不大，只不过 onErrorReturn 的范围更大一些而已。onErrorReturn 返回一个静态值，onErrorResume 返回一个动态值，并且可以捕获，包装和重新抛出错误，例如作为自定义业务异常等

onErrorResume与onErrorReturn的唯一区别是可以编写异常处理函数来返回一个给定的缺省值 * 实际上, onErrorReturn正是通过调用onErrorResume来返回缺省值 */ StepVerifier.create(generateTextFluxError() // onErrorResume还有个实用的场景是: 当发生错误时, 根据异常类型来切换到另一个流 .onErrorResume(e -> generateTextFlux().filter(s -> !Strings.isNullOrEmpty(s))) .map(s -> Integer.valueOf(s)) .reduce((x, y) -> x + y)) .expectNext(12) .verifyComplete(); }
6) 线程调度
a) 没有执行上下文(Schedulers.immediate()): 在处理时，将直接执行提交的Runnable，从而在当前线程上有效地运行它们（可以视为“空对象”或无操作调度程序）。

b) 单个可重用线程(Schedulers.single())。请注意，此方法对所有调用方都使用相同的线程，直到调度程序被释放为止。如果您需要每次调用一个专用线程，请对每个调用使用Schedulers.newSingle()。

c) 无限制的弹性线程池(Schedulers.elastic())。随着Schedulers.boundedElastic()的引入，Schedulers.boundedElastic()不再是首选方法，因为它倾向于隐藏背压问题并导致线程过多（请参见下文）。

d) 有界弹性线程池(Schedulers.boundedElastic())。像其前身elastic()一样，它根据需要创建新的工作池并重用空闲的工作池。闲置时间过长（默认值为60s）的工作池也将被丢弃。与其前身的elastic()有所不同，它对可以创建的线程数进行限制（默认为CPU核心数x 10）。达到上限后，最多可再提交10万个任务，并在有线程可用时重新调度（当任务被设置延迟执行时，延迟计时是在线程可用时开始）。这是I/O阻塞任务的更好选择。Schedulers.boundedElastic()是一种为阻塞处理提供自己的线程的简便方法，这样它就不会占用其他资源。

e) 为并行工作而调整的固定工作线程池(Schedulers.parallel())。它创建的工作线程数量与CPU内核数量一样多。

7) Mono和Flux互相转换
 // flux转mono
        Flux<String> stringFlux = Flux.fromIterable(Lists.newArrayList("a", "b2", "cc3","d4"));
        Mono<List<String>> listMono = stringFlux.collectList();
        Mono<Map<Integer, String>> mapMono = stringFlux.collectMap(String::length, String::trim);

 //mono 转flux
        Flux<String> newFlux =listMono.flatMapMany(Flux::fromIterable);
        Flux<String> newFlux2 = mapMono.map(a->{return a.values();}).flatMapMany(Flux::fromIterable);

        stringFlux.subscribe(System.out::print);
        System.out.println("---------");
        listMono.subscribe(System.out::print);
        System.out.println("---------");
        mapMono.subscribe(System.out::print);
        System.out.println("---------");
        newFlux.subscribe(System.out::print);
        System.out.println("---------");
        newFlux2.subscribe(System.out::print);

7）更多
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方法。

文章目录
什么是WebFlux 以及WebFlux 的应用场景，WebFlux【Spring  Reactive Web】
1、什么是WebFlux
2、WebFlux 的几个特征
1、异步非阻塞
2、响应式(reactive)函数编程
3、不再拘束于Servlet容器
3、WebFlux 到底有哪些优势，他是否提升了性能
4、WebFlux 应用场景
5、WebFlux 与 SpringMVC
6、WebFlux 与 SpringMVC 比较
7、WebFlux  与 Springboot

什么是WebFlux 以及WebFlux 的应用场景，WebFlux【Spring  Reactive Web】

1、什么是WebFlux
Spring WebFlux 是一个异步非阻塞式的 Web 框架，它能够充分利用多核 CPU 的硬件资源去处理大量的并发请求
他与SpringMVC 的不同

Spring MVC 构建于 Servlet API 之上，使用的是同步阻塞式 I/O 模型，什么是同步阻塞式 I/O 模型呢？就是说，每一个请求对应一个线程去处理

2、WebFlux 的几个特征
1、异步非阻塞
SpringMVC是同步阻塞的IO模型，资源浪费相对来说比较严重，当我们在处理一个比较耗时的任务时，例如：上传一个比较大的文件，首先，服务器的线程一直在等待接收文件，在这期间它就像个傻子一样等在那儿（放学别走），什么都干不了，好不容易等到文件来了并且接收完毕，我们又要将文件写入磁盘，在这写入的过程中，这根线程又再次懵bi了，又要等到文件写完才能去干其它的事情
WebFlux 就是为了解决这个问题的，WebFlux 是可以是异步非阻塞的，WebFlux 是这么处理上面的问题的

线程发现文件还没准备好，就先去做其它事情

                ↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓

当文件准备好之后，通知这根线程来处理

                ↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓

当接收完毕写入磁盘的时候（根据具体情况选择是否做异步非阻塞），

                ↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓

写入完毕后通知这根线程再来处理（异步非阻塞情况下）。
2、响应式(reactive)函数编程
得益于对于reactive-stream的支持（通过reactor框架来实现的）,WebFlux 支持函数式编程
3、不再拘束于Servlet容器
以前，我们的应用都运行于Servlet容器之中，例如我们大家最为熟悉的Tomcat, Jetty…等等。而现在Spring WebFlux不仅能运行于传统的Servlet容器中（前提是容器要支持Servlet3.1，因为非阻塞IO是使用了Servlet3.1的特性），还能运行在支持NIO的Netty和Undertow中。
3、WebFlux 到底有哪些优势，他是否提升了性能
WebFlux 内部使用的是响应式编程（Reactive Programming），以 Reactor 库为基础, 基于异步和事件驱动，可以让我们在不扩充硬件资源的前提下，提升系统的吞吐量和伸缩性
我们看看官方的原话

Reactive and non-blocking generally do not make applications run faster.

翻译出来就是 WebFlux 并不能使接口的响应时间缩短，它仅仅能够提升吞吐量和伸缩性。
4、WebFlux 应用场景
得益于 WebFlux 的异步非阻塞特性，特适合于IO 密集型的服务中，比如微服务网关

IO 密集型包括：磁盘IO密集型, 网络IO密集型，微服务网关就属于网络 IO 密集型，使用异步非阻塞式编程模型，能够显著地提升网关对下游服务转发的吞吐量。

5、WebFlux 与 SpringMVC
WebFlux 相比 SpringMVC 有一些巨大的优势，那么到底选择谁呢 ?
首先需要明确的是：WebFlux 不是 Spring MVC 的替代方案
在微服务架构中，Spring MVC 和 WebFlux 可以混合使用，比如已经提到的，对于那些 IO 密集型服务(如网关)，我们就可以使用 WebFlux 来实现
6、WebFlux 与 SpringMVC 比较
如图


相同点：
都可以使用 Spring MVC 注解，如 @Controller, 方便我们在两个 Web 框架中自由转换；

均可以使用 Tomcat, Jetty, Undertow Servlet 容器（Servlet 3.1+）；

等等
注意点：
Spring MVC 因为是使用的同步阻塞式，更方便开发人员编写功能代码，Debug 测试等，一般来说，如果 Spring MVC 能够满足的场景，就尽量不要用 WebFlux;

WebFlux 默认情况下使用 Netty 作为服务器;

WebFlux 不支持 MySql;
7、WebFlux  与 Springboot
通过IDEA  的Init 创建项目时，选择Web 里面的Spring  Reactive Web 就是 WebFlux