响应式编程系列(一):什么是响应式编程?reactor入门
2019-08-06

响应式编程 系列文章目录

(一)什么是响应式编程?reactor入门

(二)Flux入门学习:流的概念,特性和基本操作

(三)Flux深入学习:流的高级特性和进阶用法

(四)reactor-core响应式api如何测试和调试?

(五)Spring reactive: Spring WebFlux的使用

(六)Spring reactive: webClient的使用

引言

  Spring framework 5 的一大新特性:响应式编程(Reactive Programming)。那么什么是响应式?他能给我们带来什么?如何优雅地使用?本系列会从最基础的概念和简单的api讲起,再慢慢深入探讨响应式的一些高级特性,最后讲解实战内容,例如WebFlux和WebClient等在Spring boot中的使用,如何测试和调试。

  想要了解原理的话,美团点评的这篇博客 Java NIO浅析 非常适合入门。

简单地说:

  当我们调用socket.read()、socket.write()这类阻塞函数的时候,这类函数不能立即返回,也无法中断,需要等待socket可读或者可写,才会返回,因此一个线程只能处理一个请求。在这等待的过程中,cpu并不干活,(即阻塞住了),那么cpu的资源就没有很好地利用起来。因此对于这种情况,我们使用多线程来提高cpu资源的利用率:在等待的这段时间,就可以切换到别的线程去处理事件,直到socket可读或可写了,通过中断信号通知cpu,再切换回来继续处理数据。例如线程A正在等待socket可读,而线程B已经就绪了,那么就可以先切换到线程B去处理。虽然上下文切换也会花一些时间,但是远比阻塞在线程A这里空等要好。当然计算机内部实际的情况比这复杂得多。

  而NIO的读写函数可以立刻返回,这就给了我们不开线程利用CPU的最好机会:如果一个连接不能读写(socket.read()返回0或者socket.write()返回0),我们可以把这件事记下来。因此只需要一个线程不断地轮询这些事件,一旦有就绪的时间,处理即可。不需要多线程。

 

阻塞型IO

需要多线程,即需要很大的线程池。每个请求都要有一个单独的线程去处理。

 

非阻塞型IO

只需要数量非常少的线程。固定的几个工作线程去处理事件。

 

使用NIO我们能得到什么?

事件驱动模型避免多线程单线程处理多任务非阻塞I/O,I/O读写不再阻塞,而是返回0基于block的传输,通常比基于流的传输更高效更高级的IO函数,zero-copyIO多路复用大大提高了Java网络应用的可伸缩性和实用性

响应式编程入门

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

 

基本模型

我们首先需要理解响应式编程的基本模型:

 

Flux

  Reactor中的发布者(Publisher)由FluxMono两个类定义,它们都提供了丰富的操作符(operator)。一个Flux对象代表一个包含0..N个元素的响应式序列,元素可以是普通对象、数据库查询的结果、http响应体,甚至是异常。而一个Mono对象代表一个包含零/一个(0..1)元素的结果。上图就是一个Flux类型的数据流,Flux往流上发送了3个元素,Subscriber通过订阅这个流来接收通知。

如何创建一个流?最简单的方式有以下几种:

//创建一个流,并直接往流上发布一个值为value数据Flux.just(value);//通过list创建一个流,往流上依次发布list中的数据Flux.fromIterable(list);//创建一个流,并向流上从i开始连续发布n个数据,数据类型为IntegerFlux.range(i, n);//创建一个流,并定时向流上发布一个数据,数据从0开始递增,数据类型为LongFlux.interval(Duration.ofSeconds(n));

既然是“数据流”的发布者,Flux和Mono都可以发出三种“数据信号”:元素值、错误信号、完成信号,错误信号和完成信号都是终止信号,完成信号用于告知下游订阅者该数据流正常结束,错误信号终止数据流的同时将错误传递给下游订阅者。

Subscriber 

subscriber是一个订阅者,他只有非常简单的4个接口:

public interface Subscriber<T> { void onSubscribe(Subscription var1); //收到下一个元素值信号时的行为 void onNext(T var1); //收到错误信号时的行为 void onError(Throwable var1); //收到终止信号时的行为 void onComplete();}

Subscriber必须要订阅一个Flux才能够接收通知:

flux.subscribe( value -> handleData(value), error -> handleError(error), () -> handleComplete());

上面这个例子通过lambda表达式,定义了Subscriber分别在收到消息,收到错误,和消息流结束时的行为,当Subscriber接收到一个新数据,就会异步地执行handleData方法处理数据。

 

简单例子:

接下来我们创建几个最简单的流来试一下:

首先我们新建一个maven项目,引入reactor的类库:

<dependencies> <dependency> <groupId>io.projectreactor</groupId> <artifactId>reactor-core</artifactId> <version>3.2.3.RELEASE</version> </dependency> <dependency> <groupId>io.projectreactor</groupId> <artifactId>reactor-test</artifactId> <version>3.2.3.RELEASE</version> <scope>test</scope> </dependency></dependencies>

 

编写代码如下:

public class ReactorTests { @After public void after() { sleep(30_000); } @Test public void testJust() { Flux.just("hello", "world") .subscribe(System.out::println); } @Test public void testList() { List<String> words = Arrays.asList( "hello", "reactive", "world" ); Flux.fromIterable(words) .subscribe(System.out::println); } @Test public void testRange() { Flux.range(1, 10) .subscribe(System.out::println); } @Test public void testInterval() { Flux.interval(Duration.ofSeconds(1)) .subscribe(System.out::println); }}

订阅这些流,收到数据之后只是简单地把它打印出来,运行这些Test,就能够看到订阅者在接收到流上的数据时,异步地去处理这些数据。