搜索

5分钟理解SpringBoot响应式的核心-Reactor

Java3y

共 8752字，需浏览 18分钟

 ·

2019-11-07 23:23

公众号来源：美码师作者：美码师


一、前言

关于 响应式 Reactive，前面的两篇文章谈了不少概念，基本都离不开下面两点：

  • 响应式编程是面向流的、异步化的开发方式

  • 响应式是非常通用的概念，无论在前端领域、还是实时流、离线处理场景中都是适用的。

有兴趣的朋友可以看看这两篇文章：

Reactive(1) 从响应式编程到“好莱坞”

Reactive(2) 响应式流与制奶厂业务

这次，我们把目光转向 SpringBoot，在SpringBoot 2.0版本之后，提供了对响应式编程的全面支持。因此在升级到 2.x版本之后，便能方便的实现事件驱动模型的后端编程，这其中离不开 webflux这个模块。其同时也被 Spring 5 用作开发响应式 web 应用的核心基础。那么, webflux 是一个怎样的东西？

Webflux

Webflux 模块的名称是 spring-webflux，名称中的 Flux 来源于 Reactor 中的类 Flux。该模块中包含了对 响应式 HTTP、服务器推送 和 WebSocket 的支持。

Webflux 支持两种不同的编程模型：

  • 第一种是 Spring MVC 中使用的基于 Java 注解的方式，一个使用Reactive风格的Controller如下所示：
  1. @RestController
  2. publicclassEchoController {
  3. @GetMapping("/echo")
  4. publicMono<String> sayHelloWorld() {
  5. returnMono.just("Echo!");
  6.     }
  7. }
  • 第二种是 基于 Java 8 的 lambda 表达式的函数式编程模型。
这两种编程模型只是在代码编写方式上存在不同，但底层的基础模块仍然是一样的。除此之外，Webflux 可以运行在支持 Servlet 3.1 非阻塞 IO API 的 Servlet 容器上，或是其他异步运行时环境，如 Netty 和 Undertow。
关于Webflux 与 SpringMVC 的区别，可以参考下图：
4a25fd39106d3fed0d266a9b15c5ce6b.webp
SpringBoot、Webflux、Reactor 可以说是层层包含的关系，其中，响应式能力的核心仍然是来自 Reactor组件。由此可见，掌握Reactor的用法 必然是熟练进行 Spring 响应式编程的重点。
二、 Mono 与 Flux
在理解响应式Web编程之前，我们需要对Reactor 两个核心概念做一些澄清，一个是Mono，另一个是Flux。
Flux 表示的是包含 0 到 N 个元素的异步序列。在该序列中可以包含三种不同类型的消息通知：
  • 正常的包含元素的消息
  • 序列结束的消息
  • 序列出错的消息
当消息通知产生时，订阅者中对应的方法 onNext(), onComplete()和 onError()会被调用。
Mono 表示的是包含 0 或者 1 个元素的异步序列。该序列中同样可以包含与 Flux 相同的三种类型的消息通知。Flux 和 Mono 之间可以进行转换，比如对一个 Flux 序列进行计数操作，得到的结果是一个 Mono对象，或者把两个 Mono 序列合并在一起，得到的是一个 Flux 对象。
构造器
Reactor提供了非常方便的API来创建 Flux、Mono 对象，如下：
使用静态工厂类创建Flux
  1. Flux.just("Hello", "World").subscribe(System.out::println);
  2. Flux.fromArray(newInteger[] {1, 2, 3}).subscribe(System.out::println);
  3. Flux.empty().subscribe(System.out::println);
  4. Flux.range(1, 10).subscribe(System.out::println);
  5. Flux.interval(Duration.of(10, ChronoUnit.SECONDS)).subscribe(System.out::println);
  • just()：可以指定序列中包含的全部元素。创建出来的 Flux 序列在发布这些元素之后会自动结束。
  • fromArray()：可以从一个数组、Iterable 对象或 Stream 对象中创建 Flux 对象。
  • empty()：创建一个不包含任何元素，只发布结束消息的序列。
  • range(int start, int count)：创建包含从 start 起始的 count 个数量的 Integer 对象的序列。
  • interval(Duration period)和 interval(Duration delay, Duration period)：创建一个包含了从 0 开始递增的 Long 对象的序列。其中包含的元素按照指定的间隔来发布。除了间隔时间之外，还可以指定起始元素发布之前的延迟时间。
除了上述的方式之外，还可以使用 generate()、create()方法来自定义流数据的产生过程：
generate()
  1. Flux.generate(sink -> {
  2.     sink.next("Echo");
  3.     sink.complete();
  4. }).subscribe(System.out::println);
generate 只提供序列中单个消息的产生逻辑(同步通知)，其中的 sink.next()最多只能调用一次，比如上面的代码中，产生一个Echo消息后就结束了。
create()
  1. Flux.create(sink -> {
  2. for (char i = 'a'; i <= 'z'; i++) {
  3.         sink.next(i);
  4.     }
  5.     sink.complete();
  6. }).subscribe(System.out::print);
create 提供的是整个序列的产生逻辑，sink.next()可以调用多次(异步通知)，如上面的代码将会产生a-z的小写字母。
使用静态工厂类创建Mono
Mono 的创建方式与 Flux 是很相似的。除了Flux 所拥有的构造方式之外，还可以支持与Callable、Runnable、Supplier 等接口集成。
参考下面的代码：
  1. Mono.fromSupplier(() -> "Mono1").subscribe(System.out::println);
  2. Mono.justOrEmpty(Optional.of("Mono2")).subscribe(System.out::println);
  3. Mono.create(sink -> sink.success("Mono3")).subscribe(System.out::println);


三、 流计算
1. 缓冲
在Reactive(1) 从响应式编程到“好莱坞” 一文中曾经提到过缓冲(buffer)的概念。buffer 是流处理中非常常用的一种处理，意思就是将流的一段截停后再做处理。
比如下面的代码：
  1. Flux.range(1, 100).buffer(20).subscribe(System.out::println);
  2. Flux.interval(Duration.of(0, ChronoUnit.SECONDS),
  3. Duration.of(1, ChronoUnit.SECONDS))
  4.         .buffer(Duration.of(5, ChronoUnit.SECONDS)).
  5.         take(2).toStream().forEach(System.out::println);
  6. Flux.range(1, 10).bufferUntil(i -> i % 2 == 0)
  7.     .subscribe(System.out::println);
  8. Flux.range(1, 10).bufferWhile(i -> i % 2 == 0)
  9.     .subscribe(System.out::println);
第一个buffer(20)是指凑足20个数字后再进行处理，该语句会输出5组数据(按20分组)
第二个buffer(Duration duration)是指凑足一段时间后的数据再近些处理，这里是5秒钟做一次处理
第三个bufferUtil(Predicate p)是指等到某个元素满足断言(条件)时进行收集处理，这里将会输出[1,2],[3,4]..这样的奇偶数字对
第四个bufferWhile(Predicate p)则仅仅是收集满足断言(条件)的元素，这里将会输出2,4,6..这样的偶数
与 buffer 类似的是 window 函数，后者的不同在于其在缓冲截停后并不会输出一些元素列表，而是直接转换为Flux对象，如下：
  1. Flux.range(1, 100).window(20)
  2.       .subscribe(flux ->
  3.            flux.buffer(5).subscribe(System.out::println));
window(20)返回的结果是一个Flux类型的对象，我们进而对其进行了缓冲处理。因此上面的代码会按5个一组输出：
  1. [1, 2, 3, 4, 5]
  2. [6, 7, 8, 9, 10]
  3. [11, 12, 13, 14, 15]
  4. ...
2. 过滤/提取
上面的bufferWhile 其实充当了过滤的作用，当然，对于流元素的过滤也可以使用 filter函数来处理：
  1. Flux.range(1, 10).filter(i -> i % 2 == 0).subscribe(System.out::println);
take 函数 可以用来提取想要的元素，这与filter 过滤动作是恰恰相反的，来看看take的用法：
  1. Flux.range(1, 10).take(2).subscribe(System.out::println);
  2. Flux.range(1, 10).takeLast(2).subscribe(System.out::println);
  3. Flux.range(1, 10).takeWhile(i -> i < 5).subscribe(System.out::println);
  4. Flux.range(1, 10).takeUntil(i -> i == 6).subscribe(System.out::println);
第一个take(2)指提取前面的两个元素；第二个takeLast(2)指提取最后的两个元素；第三个takeWhile(Predicate p)指提取满足条件的元素，这里是1-4
第四个takeUtil(Predicate p)指一直提取直到满足条件的元素出现为止，这里是1-6
3. 转换
使用map函数可以将流中的元素进行个体转换，如下：
  1. Flux.range(1, 10).map(x -> x*x).subscribe(System.out::println);
这里的map使用的JDK8 所定义的 Function接口
4. 合并
某些情况下我们需要对两个流中的元素进行合并处理，这与合并两个数组有点相似，但结合流的特点又会有不同的需求。
使用 zipWith 函数可以实现简单的流元素合并处理：
  1. Flux.just("I", "You")
  2.         .zipWith(Flux.just("Win", "Lose"))
  3.         .subscribe(System.out::println);
  4. Flux.just("I", "You")
  5.         .zipWith(Flux.just("Win", "Lose"),
  6.         (s1, s2) -> String.format("%s!%s!", s1, s2))
  7.         .subscribe(System.out::println);
上面的代码输出为：
  1. [I,Win]
  2. [You,Lose]
  3. I!Win!
  4. You!Lose!
第一个zipWith输出的是Tuple对象(不可变的元祖)，第二个zipWith增加了一个BiFunction来实现合并计算，输出的是字符串。
注意到zipWith是分别按照元素在流中的顺序进行两两合并的，合并后的流长度则最短的流为准，遵循最短对齐原则。
用于实现合并的还有 combineLastest函数，combinLastest 会动态的将流中新产生元素(末位)进行合并，注意是只要产生新元素都会触发合并动作并产生一个结果元素，如下面的代码：
  1. Flux.combineLatest(
  2. Arrays::toString,
  3. Flux.interval(Duration.of(0, ChronoUnit.MILLIS),
  4. Duration.of(100, ChronoUnit.MILLIS)).take(2),
  5. Flux.interval(Duration.of(50, ChronoUnit.MILLIS),
  6. Duration.of(100, ChronoUnit.MILLIS)).take(2)
  7. ).toStream().forEach(System.out::println);
输出为：
  1. [0, 0]
  2. [1, 0]
  3. [1, 1]
5. 合流
与合并比较类似的处理概念是合流，合流的不同之处就在于元素之间不会产生合并，最终流的元素个数(长度)是两个源的个数之和。合流的计算可以使用 merge或mergeSequential 函数，这两者的区别在于：
merge后的元素是按产生时间排序的，而mergeSequential 则是按整个流被订阅的时间来排序，如下面的代码：
  1. Flux.merge(Flux.interval(
  2. Duration.of(0, ChronoUnit.MILLIS),
  3. Duration.of(100, ChronoUnit.MILLIS)).take(2),
  4. Flux.interval(
  5. Duration.of(50, ChronoUnit.MILLIS),
  6. Duration.of(100, ChronoUnit.MILLIS)).take(2))
  7.         .toStream()
  8.         .forEach(System.out::println);
  9. System.out.println("---");
  10. Flux.mergeSequential(Flux.interval(
  11. Duration.of(0, ChronoUnit.MILLIS),
  12. Duration.of(100, ChronoUnit.MILLIS)).take(2),
  13. Flux.interval(
  14. Duration.of(50, ChronoUnit.MILLIS),
  15. Duration.of(100, ChronoUnit.MILLIS)).take(2))
  16.         .toStream()
  17.         .forEach(System.out::println);
输出为：
  1. 0
  2. 0
  3. 1
  4. 1
  5. ---
  6. 0
  7. 1
  8. 0
  9. 1
merge 是直接将Flux 元素进行合流之外，而flatMap则提供了更加高级的处理：flatMap 函数会先将Flux中的元素转换为 Flux(流)，然后再新产生的Flux进行合流处理，如下：
  1. Flux.just(1, 2)
  2.         .flatMap(x -> Flux.interval(
  3. Duration.of(x * 10, ChronoUnit.MILLIS),
  4. Duration.of(10, ChronoUnit.MILLIS)).take(x))
  5.         .toStream()
  6.         .forEach(System.out::println);
flatMap也存在flatMapSequential的一个兄弟版本，后者决定了合并流元素的顺序是与流的订阅顺序一致的。
6. 累积
reduce 和 reduceWith 操作符对流中包含的所有元素进行累积操作，得到一个包含计算结果的 Mono 序列。累积操作是通过一个 BiFunction 来表示的。reduceWith 允许在在操作时指定一个起始值(与第一个元素进行运算)
如下面的代码：
  1. Flux.range(1, 100).reduce((x, y) -> x + y)
  2.     .subscribe(System.out::println);
  3. Flux.range(1, 100).reduceWith(() -> 100, (x, y) -> x + y)
  4.     .subscribe(System.out::println);
这里通过reduce计算出1-100的累加结果(1+2+3+...100)，结果输出为：
  1. 5050
  2. 5150

四、异常处理
在前面所提及的这些功能基本都属于正常的流处理，然而对于异常的捕获以及采取一些修正手段也是同样重要的。
利用Flux/Mono 框架可以很方便的做到这点。
将正常消息和错误消息分别打印
  1. Flux.just(1, 2)
  2.         .concatWith(Mono.error(newIllegalStateException()))
  3.         .subscribe(System.out::println, System.err::println);
当产生错误时默认返回0
  1. Flux.just(1, 2)
  2.         .concatWith(Mono.error(newIllegalStateException()))
  3.         .onErrorReturn(0)
  4.         .subscribe(System.out::println);
自定义异常时的处理
  1. Flux.just(1, 2)
  2.         .concatWith(Mono.error(newIllegalArgumentException()))
  3.         .onErrorResume(e -> {
  4. if (e instanceofIllegalStateException) {
  5. returnMono.just(0);
  6.             } elseif (e instanceofIllegalArgumentException) {
  7. returnMono.just(-1);
  8.             }
  9. returnMono.empty();
  10.         })
  11.         .subscribe(System.out::println);
当产生错误时重试
  1. Flux.just(1, 2)
  2.         .concatWith(Mono.error(newIllegalStateException()))
  3.         .retry(1)
  4.         .subscribe(System.out::println);
这里的retry(1)表示最多重试1次，而且重试将从订阅的位置开始重新发送流事件

五、线程调度
我们说过，响应式是异步化的，那么就会涉及到多线程的调度。
Reactor 提供了非常方便的调度器(Scheduler)工具方法，可以指定流的产生以及转换(计算)发布所采用的线程调度方式。这些方式包括：
类别描述
immediate采用当前线程
single单一可复用的线程
elastic弹性可复用的线程池(IO型)
parallel并行操作优化的线程池(CPU计算型)
timer支持任务调度的线程池
fromExecutorService自定义线程池
下面，以一个简单的实例来演示不同的线程调度：
  1. Flux.create(sink -> {
  2.     sink.next(Thread.currentThread().getName());
  3.     sink.complete();
  4. })
  5. .publishOn(Schedulers.single())
  6. .map(x -> String.format("[%s] %s", Thread.currentThread().getName(), x))
  7. .publishOn(Schedulers.elastic())
  8. .map(x -> String.format("[%s] %s", Thread.currentThread().getName(), x))
  9. .subscribeOn(Schedulers.parallel())
  10. .toStream()
  11. .forEach(System.out::println);
在这段代码中，使用 publishOn 指定了流发布的调度器，subscribeOn 则指定的是流产生的调度器。
首先是parallel调度器进行流数据的生成，接着使用一个single单线程调度器进行发布，此时经过第一个map转换为另一个Flux流，其中的消息叠加了当前线程的名称。最后进入的是一个elastic弹性调度器，再次进行一次同样的map转换。
最终，经过多层转换后的输出如下：
  1. [elastic-2] [single-1] parallel-1

小结
SpringBoot 2.x、Spring 5 对于响应式的Web编程(基于Reactor)都提供了全面的支持，借助于框架的能力可以快速的完成一些简单的响应式代码开发。本文提供了较多 Reactor API的代码样例，旨在帮助读者能快速的理解 响应式编程的概念及方式。
对于习惯了传统编程范式的开发人员来说，熟练使用 Reactor 仍然需要一些思维上的转变。就笔者的自身感觉来看，Reactor 存在一些学习和适应的成本，但一旦熟悉使用之后便能体会它的先进之处。就如 JDK8 引入的Stream API之后，许多开发者则渐渐抛弃forEach的方式..
本身这就是一种生产效率的提升机会，何乐而不为？更何况，从应用框架的发展前景来看，响应式的前景是明朗的。
两年呕心沥血的文章：「面试题」「基础」「进阶」这里全都有！

200多篇原创技术文章海量视频资源精美脑图面试题
长按扫码可关注获取 
在看和分享对我非常重要！51e9c1156ddeffbda3551a7c14e9efae.webp

近期推荐：低价购买云服务器+搭建教程
浏览
                    45
点赞
    
评论
    
收藏
    
分享
        
手机扫一扫分享
分享
    

        举报
    
评论
图片
表情
推荐
        
Project Reactor 响应式编程
肥宅与画家
0
手把手带你理解Vue响应式原理
web前端开发
0
SpringBoot AngularJS FileManager响应式文件管理器
基于SpringBoot、AngularJS、CSS3、HTML5的响应式文件浏览管理器，spring-boot-filemanager 方便有文件管理功能的项目集成,不依赖后端,目前只集成了Spri
五分钟快速理解 Reactor 模型
Kirito的技术分享
0
ReactKitSwift 的响应式编程
ReactKit是Swift的响应式编程。示例代码：// create signal via KVOself.obj1Signal = KVO.signal(obj1, "value")// bind
ReactKitSwift 的响应式编程
ReactKit 是 Swift 的响应式编程。示例代码：// create signal via 
响应式网页
 一个网站所有的样式全部写在一个css文件中，写类名把父元素带上，这样就不会有类名冲突了响应式就是使用媒体查询，在多个的屏幕尺寸下，让网页的布局不要乱。核心就是，在对应的屏幕下，通过对发生错乱的元素样式进...
springboot 源码理解
java1234
0
设计体系的响应式设计
（给前端大学加星标，提升前端技能.）作者：TChttps://zhuanlan.zhihu.com/p/109781992导读在蚂蚁内部有着数量繁多且复杂的中后台业务系统，Ant Design 一直以来致力于从设计策略和资产的角度解决这些产品的体验一致...
MQFramework响应式的 CSS 框架
MQFramework 是一个响应式的 CSS 框架，利用 @media 属性来适应不同的浏览器和尺
点赞
    
评论
    
收藏
    
分享
        
手机扫一扫分享
分享
    

        举报