5分钟理解 SpringBoot 响应式的核心-Reactor
Posted littleatp
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了5分钟理解 SpringBoot 响应式的核心-Reactor相关的知识,希望对你有一定的参考价值。
一、前言
关于 响应式 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如下所示:
@RestController
public class EchoController
@GetMapping("/echo")
public Mono<String> sayHelloWorld()
return Mono.just("Echo!");
- 第二种是 基于 Java 8 的 lambda 表达式的函数式编程模型。
这两种编程模型只是在代码编写方式上存在不同,但底层的基础模块仍然是一样的。
除此之外,Webflux 可以运行在支持 Servlet 3.1 非阻塞 IO API 的 Servlet 容器上,或是其他异步运行时环境,如 Netty 和 Undertow。
关于Webflux 与 SpringMVC 的区别,可以参考下图:
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
构造器
Reactor提供了非常方便的API来创建 Flux、Mono 对象,如下:
使用静态工厂类创建Flux
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);
- 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()
Flux.generate(sink ->
sink.next("Echo");
sink.complete();
).subscribe(System.out::println);
generate 只提供序列中单个消息的产生逻辑(同步通知),其中的 sink.next()最多只能调用一次,比如上面的代码中,产生一个Echo消息后就结束了。
create()
Flux.create(sink ->
for (char i = 'a'; i <= 'z'; i++)
sink.next(i);
sink.complete();
).subscribe(System.out::print);
create 提供的是整个序列的产生逻辑,sink.next()可以调用多次(异步通知),如上面的代码将会产生a-z的小写字母。
使用静态工厂类创建Mono
Mono 的创建方式与 Flux 是很相似的。 除了Flux 所拥有的构造方式之外,还可以支持与Callable、Runnable、Supplier 等接口集成。
参考下面的代码:
Mono.fromSupplier(() -> "Mono1").subscribe(System.out::println);
Mono.justOrEmpty(Optional.of("Mono2")).subscribe(System.out::println);
Mono.create(sink -> sink.success("Mono3")).subscribe(System.out::println);
三、 流计算
1. 缓冲
在Reactive(1) 从响应式编程到“好莱坞” 一文中曾经提到过缓冲(buffer)的概念。
buffer 是流处理中非常常用的一种处理,意思就是将流的一段截停后再做处理。
比如下面的代码:
Flux.range(1, 100).buffer(20).subscribe(System.out::println);
Flux.interval(Duration.of(0, ChronoUnit.SECONDS),
Duration.of(1, ChronoUnit.SECONDS))
.buffer(Duration.of(5, ChronoUnit.SECONDS)).
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);
第一个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对象,如下:
Flux.range(1, 100).window(20)
.subscribe(flux ->
flux.buffer(5).subscribe(System.out::println));
window(20)返回的结果是一个Flux
因此上面的代码会按5个一组输出:
[1, 2, 3, 4, 5]
[6, 7, 8, 9, 10]
[11, 12, 13, 14, 15]
...
2. 过滤/提取
上面的bufferWhile 其实充当了过滤的作用,当然,对于流元素的过滤也可以使用filter函数来处理:
Flux.range(1, 10).filter(i -> i % 2 == 0).subscribe(System.out::println);
take 函数可以用来提取想要的元素,这与filter 过滤动作是恰恰相反的,来看看take的用法:
Flux.range(1, 10).take(2).subscribe(System.out::println);
Flux.range(1, 10).takeLast(2).subscribe(System.out::println);
Flux.range(1, 10).takeWhile(i -> i < 5).subscribe(System.out::println);
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函数可以将流中的元素进行个体转换,如下:
Flux.range(1, 10).map(x -> x*x).subscribe(System.out::println);
这里的map使用的JDK8 所定义的 Function接口
4. 合并
某些情况下我们需要对两个流中的元素进行合并处理,这与合并两个数组有点相似,但结合流的特点又会有不同的需求。
使用zipWith函数可以实现简单的流元素合并处理:
Flux.just("I", "You")
.zipWith(Flux.just("Win", "Lose"))
.subscribe(System.out::println);
Flux.just("I", "You")
.zipWith(Flux.just("Win", "Lose"),
(s1, s2) -> String.format("%s!%s!", s1, s2))
.subscribe(System.out::println);
上面的代码输出为:
[I,Win]
[You,Lose]
I!Win!
You!Lose!
第一个zipWith输出的是Tuple对象(不可变的元祖),第二个zipWith增加了一个BiFunction来实现合并计算,输出的是字符串。
注意到zipWith是分别按照元素在流中的顺序进行两两合并的,合并后的流长度则最短的流为准,遵循最短对齐原则。
用于实现合并的还有 combineLastest函数,combinLastest 会动态的将流中新产生元素(末位)进行合并,注意是只要产生新元素都会触发合并动作并产生一个结果元素,如下面的代码:
Flux.combineLatest(
Arrays::toString,
Flux.interval(Duration.of(0, ChronoUnit.MILLIS),
Duration.of(100, ChronoUnit.MILLIS)).take(2),
Flux.interval(Duration.of(50, ChronoUnit.MILLIS),
Duration.of(100, ChronoUnit.MILLIS)).take(2)
).toStream().forEach(System.out::println);
输出为:
[0, 0]
[1, 0]
[1, 1]
5. 合流
与合并比较类似的处理概念是合流,合流的不同之处就在于元素之间不会产生合并,最终流的元素个数(长度)是两个源的个数之和。
合流的计算可以使用 merge或mergeSequential 函数,这两者的区别在于:
merge后的元素是按产生时间排序的,而mergeSequential 则是按整个流被订阅的时间来排序,如下面的代码:
Flux.merge(Flux.interval(
Duration.of(0, ChronoUnit.MILLIS),
Duration.of(100, ChronoUnit.MILLIS)).take(2),
Flux.interval(
Duration.of(50, ChronoUnit.MILLIS),
Duration.of(100, ChronoUnit.MILLIS)).take(2))
.toStream()
.forEach(System.out::println);
System.out.println("---");
Flux.mergeSequential(Flux.interval(
Duration.of(0, ChronoUnit.MILLIS),
Duration.of(100, ChronoUnit.MILLIS)).take(2),
Flux.interval(
Duration.of(50, ChronoUnit.MILLIS),
Duration.of(100, ChronoUnit.MILLIS)).take(2))
.toStream()
.forEach(System.out::println);
输出为:
0
0
1
1
---
0
1
0
1
merge 是直接将Flux 元素进行合流之外,而flatMap则提供了更加高级的处理:
flatMap 函数会先将Flux中的元素转换为 Flux(流),然后再新产生的Flux进行合流处理,如下:
Flux.just(1, 2)
.flatMap(x -> Flux.interval(Duration.of(x * 10, ChronoUnit.MILLIS),
Duration.of(10, ChronoUnit.MILLIS)).take(x))
.toStream()
.forEach(System.out::println);
flatMap也存在flatMapSequential的一个兄弟版本,后者决定了合并流元素的顺序是与流的订阅顺序一致的。
6. 累积
reduce 和 reduceWith 操作符对流中包含的所有元素进行累积操作,得到一个包含计算结果的 Mono 序列。累积操作是通过一个 BiFunction 来表示的。reduceWith 允许在在操作时指定一个起始值(与第一个元素进行运算)
如下面的代码:
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);
这里通过reduce计算出1-100的累加结果(1+2+3+...100),结果输出为:
5050
5150
四、异常处理
在前面所提及的这些功能基本都属于正常的流处理,然而对于异常的捕获以及采取一些修正手段也是同样重要的。
利用Flux/Mono 框架可以很方便的做到这点。
将正常消息和错误消息分别打印
Flux.just(1, 2)
.concatWith(Mono.error(new IllegalStateException()))
.subscribe(System.out::println, System.err::println);
当产生错误时默认返回0
Flux.just(1, 2)
.concatWith(Mono.error(new IllegalStateException()))
.onErrorReturn(0)
.subscribe(System.out::println);
自定义异常时的处理
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);
当产生错误时重试
Flux.just(1, 2)
.concatWith(Mono.error(new IllegalStateException()))
.retry(1)
.subscribe(System.out::println);
这里的retry(1)表示最多重试1次,而且重试将从订阅的位置开始重新发送流事件
五、线程调度
我们说过,响应式是异步化的,那么就会涉及到多线程的调度。
Reactor 提供了非常方便的调度器(Scheduler)工具方法,可以指定流的产生以及转换(计算)发布所采用的线程调度方式。
这些方式包括:
类别 | 描述 |
---|---|
immediate | 采用当前线程 |
single | 单一可复用的线程 |
elastic | 弹性可复用的线程池(IO型) |
parallel | 并行操作优化的线程池(CPU计算型) |
timer | 支持任务调度的线程池 |
fromExecutorService | 自定义线程池 |
下面,以一个简单的实例来演示不同的线程调度:
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);
在这段代码中,使用publishOn指定了流发布的调度器,subscribeOn则指定的是流产生的调度器。
首先是parallel调度器进行流数据的生成,接着使用一个single单线程调度器进行发布,此时经过第一个map转换为另一个Flux流,其中的消息叠加了当前线程的名称。最后进入的是一个elastic弹性调度器,再次进行一次同样的map转换。
最终,经过多层转换后的输出如下:
[elastic-2] [single-1] parallel-1
小结
SpringBoot 2.x、Spring 5 对于响应式的Web编程(基于Reactor)都提供了全面的支持,借助于框架的能力可以快速的完成一些简单的响应式代码开发。
本文提供了较多 Reactor API的代码样例,旨在帮助读者能快速的理解 响应式编程的概念及方式。
对于习惯了传统编程范式的开发人员来说,熟练使用 Reactor 仍然需要一些思维上的转变。
就笔者的自身感觉来看,Reactor 存在一些学习和适应的成本,但一旦熟悉使用之后便能体会它的先进之处。 就如 JDK8 引入的Stream API之后,许多开发者则渐渐抛弃forEach的方式..
本身这就是一种生产效率的提升机会,何乐而不为? 更何况,从应用框架的发展前景来看,响应式的前景是明朗的。
参考阅读
使用 Reactor 进行反应式编程
https://www.ibm.com/developerworks/cn/java/j-cn-with-reactor-response-encode/index.html
Spring 5 的 WebFlux 开发介绍
https://www.ibm.com/developerworks/cn/java/spring5-webflux-reactive/index.html
以上是关于5分钟理解 SpringBoot 响应式的核心-Reactor的主要内容,如果未能解决你的问题,请参考以下文章