杜万专家深度解析：Java响应式编程全攻略

摘要： 响应式宣言如何解读，Java中如何进行响应式编程，Reactor Streams又该如何使用？热衷于整合框架与开发工具的阿里云技术专家杜万，为大家全面解读响应式编程，分享Spring Webflux的实践。

本篇文章来自于2018年12月22日举办的《阿里云栖开发者沙龙—Java技术专场》，杜万专家是该专场第四位演讲的嘉宾，本篇文章是根据杜万专家在《阿里云栖开发者沙龙—Java技术专场》的演讲视频以及PPT整理而成。

摘要：响应式宣言如何解读，Java中如何进行响应式编程，Reactor Streams又该如何使用？热衷于整合框架与开发工具的阿里云技术专家杜万，为大家全面解读响应式编程，分享Spring Webflux的实践。从响应式理解，到Reactor项目示例，再到Spring Webflux框架解读，本文带你进入Java响应式编程。

演讲嘉宾简介：

杜万（倚贤），阿里云技术专家，全栈工程师，从事了12年 Java 语言为主的软件开发工作，热衷于整合框架与开发工具，Linux拥趸，问题终结者。合作翻译《Elixir 程序设计》。目前负责阿里云函数计算的工具链开发，正在实践 WebFlux 和 Reactor 开发新的 Web 应用。

本次直播视频精彩回顾，戳这里！
https://yq.aliyun.com/live/721

PPT下载地址：
https://yq.aliyun.com/download/3187

以下内容根据演讲嘉宾视频分享以及PPT整理而成。

本文围绕以下三部分进行介绍：

1.Reactive

2.Project Reactor

3.Spring Webflux

一．Reactive

1.Reactive Manifesto

下图是Reactive Manifesto官方网站上的介绍，这篇文章非常短但也非常精悍，非常值得大家去认真阅读。

响应式宣言是一份构建现代云扩展架构的处方。这个框架主要使用消息驱动的方法来构建系统，在形式上可以达到弹性和韧性，最后可以产生响应性的价值。所谓弹性和韧性，通俗来说就像是橡皮筋，弹性是指橡皮筋可以拉长，而韧性指在拉长后可以缩回原样。这里为大家一一解读其中的关键词：

1）响应性：快速/一致的响应时间。假设在有500个并发操作时，响应时间为1s，那么并发操作增长至5万时，响应时间也应控制在1s左右。快速一致的响应时间才能给予用户信心，是系统设计的追求。

2）韧性：复制/遏制/隔绝/委托。当某个模块出现问题时，需要将这个问题控制在一定范围内，这便需要使用隔绝的技术，避免连锁性问题的发生。或是将出现故障部分的任务委托给其他模块。韧性主要是系统对错误的容忍。

3）弹性：无竞争点或中心瓶颈/分片/扩展。如果没有状态的话，就进行水平扩展，如果存在状态，就使用分片技术，将数据分至不同的机器上。

4）消息驱动：异步/松耦合/隔绝/地址透明/错误作为消息/背压/无阻塞。消息驱动是实现上述三项的技术支撑。其中，地址透明有很多方法。例如DNS提供的一串人类能读懂的地址，而不是IP，这是一种不依赖于实现，而依赖于声明的设计。再例如k8s每个service后会有多个Pod，依赖一个虚拟的服务而不是某一个真实的实例，从何实现调用1 个或调用n个服务实例对于对调用方无感知，这是为分片或扩展做了准备。错误作为消息，这在Java中是不太常见的，Java中通常将错误直接作为异常抛出，而在响应式中，错误也是一种消息，和普通消息地位一致，这和JavaScript中的Promise类似。背压是指当上游向下游推送数据时，可能下游承受能力不足导致问题，一个经典的比喻是就像用消防水龙头解渴。因此下游需要向上游声明每次只能接受大约多少量的数据，当接受完毕再次向上游申请数据传输。这便转换成是下游向上游申请数据，而不是上游向下游推送数据。无阻塞是通过no-blocking IO提供更高的多线程切换效率。

2.Reactive Programming

响应式编程是一种声明式编程范型。下图中左侧显示了一个命令式编程，相信大家都比较熟悉。先声明两个变量，然后进行赋值，让两个变量相加，得到相加的结果。但接着当修改了最早声明的两个变量的值后，sum的值不会因此产生变化。而在Java 9 Flow中，按相同的思路实现上述处理流程，当初始变量的值变化，最后结果的值也同步发生变化，这就是响应式编程。这相当于声明了一个公式，输出值会随着输入值而同步变化。

响应式编程也是一种非阻塞的异步编程。下图是用reactor.ipc.netty实现的TCP通信。常见的server中会用循环发数据后，在循环外取出，但在下图的实现中没有，因为这不是使用阻塞模型实现，是基于非阻塞的异步编程实现。

响应式编程是一种数据流编程，关注于数据流而不是控制流。下图中，首先当页面出现点击操作时产生一个click stream，然后页面会将250ms内的clickStream缓存，如此实现了一个归组过程。然后再进行map操作，得到每个list的长度，筛选出长度大于2的，这便可以得出多次点击操作的流。这种方法应用非常广泛，例如可以筛选出双击操作。由此可见，这种编程方式是一种数据流编程，而不是if else的控制流编程。

之前有提及消息驱动，那么消息驱动（Message-driven）和事件驱动（Event-driven）有什么区别呢。

1）消息驱动有确定的目标，一定会有消息的接受者，而事件驱动是一件事情希望被观察到，观察者是谁无关紧要。消息驱动系统关注消息的接受者，事件驱动系统关注事件源。

2）在一个使用响应式编程实现的响应式系统中，消息擅长于通讯，事件擅长于反应事实。

3.Reactive Streams

Reactive Streams提供了一套非阻塞背压的异步流处理标准，主要应用在JVM、JavaScript和网络协议工作中。通俗来说，它定义了一套响应式编程的标准。在Java中，有4个Reactive Streams API，如下图所示：

这个API中定义了Publisher，即事件的发生源，它只有一个subscribe方法。其中的Subscriber就是订阅消息的对象。

作为订阅者，有四个方法。onSubscribe会在每次接收消息时调用，得到的数据都会经过onNext方法。onError方法会在出现问题时调用，Throwable即是出现的错误消息。在结束时调用onComplete方法。

Subscription接口用来描述每个订阅的消息。request方法用来向上游索要指定个数的消息，cancel方法用于取消上游的数据推送，不再接受消息。

Processor接口继承了Subscriber和Publisher，它既是消息的发生者也是消息的订阅者。这是发生者和订阅者间的过渡桥梁，负责一些中间转换的处理。

Reactor Library从开始到现在已经历经多代。第0代就是java包Observable 接口，也就是观察者模式。具体的发展见下图：

第四代虽然仍然是RxJava2，但是相比第三代的RxJava2，其中的小版本有了不一样的改进，出现了新特性。

Reactor Library主要有两点特性。一是基于回调（callback-based），在事件源附加回调函数，并在事件通过数据流链时被调用；二是声明式编程（Declarative），很多函数处理业务类似，例如map/filter/fold等，这些操作被类库固化后便可以使用声明式方法，以在程序中快速便捷使用。在生产者、订阅者都定义后，声明式方法便可以用来实现中间处理者。

二．Project Reactor

Project Reactor，实现了完全非阻塞，并且基于网络HTTP/TCP/UDP等的背压，即数据传输上游为网络层协议时，通过远程调用也可以实现背压。同时，它还实现了Reactive Streams API和Reactive Extensions，以及支持Java 8 functional API/Completable Future/Stream /Duration等各新特性。下图所示为Reactor的一个示例：

首先定义了一个words的数组，然后使用flatMap做映射，再将每个词和s做连接，得出的结果和另一个等长的序列进行一个zipWith操作，最后打印结果。这和Java 8 Stream非常类似，但仍存在一些区别：

1）Stream是pull-based，下游从上游拉数据的过程，它会有中间操作例如map和reduce，和终止操作例如collect等，只有在终止操作时才会真正的拉取数据。Reactive是push-based，可以先将整个处理数据量构造完成，然后向其中填充数据，在出口处可以取出转换结果。

2）Stream只能使用一次，因为它是pull-based操作，拉取一次之后源头不能更改。但Reactive可以使用多次，因为push-based操作像是一个数据加工厂，只要填充数据就可以一直产出。

3）Stream#parallel()使用fork-join并发，就是将每一个大任务一直拆分至指定大小颗粒的小任务，每个小任务可以在不同的线程中执行，这种多线程模型符合了它的多核特性。Reactive使用Event loop，用一个单线程不停的做循环，每个循环处理有限的数据直至处理完成。

在上例中，大家可以看到很多Reactive的操作符，例如
flatMap/concatWith/zipWith等，这样的操作符有300多个，这可能是学习这个框架最大的压力。如何理解如此繁多的操作符，可能一个归类会有所帮助：

1）新序列创建，例如创建数组类序列等；

2）现有序列转换，将其转换为新的序列，例如常见的map操作；

3）从现有的序列取出某些元素；

4）序列过滤；

5）序列异常处理。

6）与时间相关的操作，例如某个序列是由时间触发器定期发起事件；

7）序列分割；

8）序列拉至同步世界，不是所有的框架都支持异步，再需要和同步操作进行交互时就需要这种处理。

上述300+操作符都有如下所示的弹珠图（Marble Diagrams），用表意的方式解释其作用。例如下图的操作符是指，随着时间推移，逐个产生了6个元素的序列，黑色竖线表示新元素产生终止。在这个操作符的作用下，下方只取了前三个元素，到第四个元素就不取了。这些弹珠图大家可以自行了解。

三．Spring Webflux

1.Spring Webflux框架

Spring Boot 2.0相较之前的版本，在基于Spring Framework 5的构建添加了新模块Webflux，将默认的web服务器改为Netty，支持Reactive应用，并且Webflux默认运行在Netty上。而Spring Framework 5也有了一些变化。Java版本最低依赖Java 8，支持Java 9和Java 10，提供许多支持Reactive的基础设施，提供面向Netty等运行时环境的适配器，新增Webflux模块（集成的是Reactor 3.x）。下图所示为Webflux的框架：

左侧是通常使用的框架，通过Servlet API的规范和Container进行交互，上一层是Spring-Webmvc，再上一层则是经常使用的一些注解。右侧为对应的Webflux层级，只要是支持NIO的Container，例如Tomcat，Jetty，Netty或Undertow都可以实现。在协议层的是HTTP/Reactive Streams。再上一层是Spring-Webflux，为了保持兼容性，它支持这些常用的注解，同时也有一套新的语法规则Router Functions。下图显示了一个调用的实例：

在Client端，首先创建一个WebClient，调用其get方法，写入URL，接收格式为APPLICATION_STREAM_JSON的数据，retrieve获得数据，取得数据后用bodyToFlux将数据转换为Car类型的对象，在doOnNext中打印构造好的Car对象，block方法意思是直到回调函数被执行才可以结束。在Server端，在指定的path中进行get操作，produces和以前不同，这里是application/stream+json，然后返回Flux范型的Car对象。传统意义上，如果数据中有一万条数据，那么便直接返回一万条数据，但在这个示例返回的Flux范型中，是不包含数据的，但在数据库也支持Reactive的情况下，request可以一直往下传递，响应式的批量返回。传统方式这样的查询很有可能是一个全表遍历，这会需要较多资源和时间，甚至影响其他任务的执行。而响应式的方法除了可以避免这种情况，还可以让用户在第一时间看到数据而不是等待数据采集完毕，这在架构体验的完整性上有了很大的提升。application/stream+json也是可以让前端识别出，这些数据是分批响应式传递，而不会等待传完才显示。

现在的Java web应用可以使用Servlet栈或Reactive栈。Servlet栈已经有很久的使用历史了，而现在又增加了更有优势的Reactive栈，大家可以尝试实现更好的用户体验。

2.Reactive编程模型

下图中是Spring实现的一个向后兼容模型，可以使用annotation来标注Container。这是一个非常清晰、支持非常细节化的模型，也非常利于同事间的交流沟通。

下图是一个Functional编程模型，通过写函数的方式构造。例如下图中传入一个Request，返回Response，通过函数的方法重点关注输入输出，不需要区分状态。然后将这些函数注册至Route。这个模型和Node.js非常接近，也利于使用。

3.Spring Data框架

Spring Data框架支持多种数据库，如下图所示，最常用的是JPA和JDBC。在实践中，不同的语言访问不同的数据库时，访问接口是不一样的，这对编程人员来说是个很大的工作量。

Spring Data便是做了另一层抽象，使你无论使用哪种数据库，都可以使用同一个接口。具体特性这里不做详谈。

下图展示了一个Spring Data的使用示例。只需要写一个方法签名，然后注解为Query，这个方法不需要实现，因为框架后台已经采用一些技术，直接根据
findByFirstnameAndLastname就可以查询到。这种一致的调用方式无疑提供了巨大的方便。

现在Reactive对Spring Data的支持还是不完整的，只支持了MongoDB/Redis/Cassandra和Couchbase，对
JPA/LDAP/Elasticsearch/Neo4j/Solr等还不兼容。但也不是不能使用，例如对JDBC数据库，将其转为同步即可使用，重点在于findAll和async两个函数，这里不再展开详述，具体代码如下图所示：

Reactive不支持JDBC最根本的原因是，JDBC不是non-blocking设计。但是现在JavaOne已经在2016年9月宣布了Non-blocking JDBC API的草案，虽然还未得到Java 10的支持，但可见这已经成为一种趋势。

四．总结

Spring MVC框架是一个命令式逻辑，方便编写和调试。Spring WebFlux也具有众多优势，但调试却不太容易，因为它经常需要切换线程执行，出现错误的栈可能已经销毁。当然这也是现今Java的编译工具对WebFlux不太友好，相信以后会改善。下图中列出了Spring MVC和Spring WebFlux各自的特性及交叉的部分。最后也附上一些参考资料。

作者：李博bluemind