响应式编程介绍

怼烎@ 2023-09-23 21:59 131阅读 0赞

## 1、响应式宣言 ##

说起响应式，就不得不先说下响应式宣言-[Reactive Manifesto][]

在响应式宣言中阐述了『响应式系统』该有的特质以及实现手段：

![format_png][]

总结来说 就是 一个响应式的系统，需要具备在服务正常或者异常的情况下，都需要及时的对外部请求做出响应，响应可以是正常内容也可以是异常情况下的 fast throw，借助于消息驱动或者事件驱动来达到系统内部异步非阻塞的交互机制，这通常需要系统服务各个分层各个组件都是响应式的。

其中 即时响应和消息驱动是侧重点，实现消息驱动的方式非常多，在这样的背景下需要有一套实现规范，所有的响应式编程框架都应该按照规范进行设计与实现

## 2、响应式编程 ##

响应式编程的实现，基本是以 响应式标准规范实现的，当然也有些例外，比如 akka stream。

### 2.1、Reactive Stream ###

Reactive Stream，响应式规范，即响应式流，是一个异步Stream 的响应式规范标准，它定义了实现响应式编程时的 API（类似于 JDBC 标准 API），囊括多种语言，其中 JVM 平台的API 定义见 [reactive-streams-jvm][]

在 API 中可以看到有四种角色：

Publisher：事件发布者

![format_png 1][]

Subscriber: 事件订阅消费者，定义了消费的一般模式，onSubscribe onNext\* (onError | onComplete)?

![format_png 2][]

Subscrition： 表示一次消息生产到消费的关系，是一个消息的声明周期

![format_png 3][]

Processor：从类继承来看，它既是订阅者也是发布者，表示一个处理的阶段

![format_png 4][]

从这些角色也可以看到，发布者订阅者，这和我们常见的消息中间件 以及 异动任务处理的方式都非常像。

![format_png 5][]

![format_png 6][]

### 2.2、响应式编程 ###

响应式编程的核心思想 是在 数据流 上，一切皆为流，数据流的变化传递体现在数据流处理节点流转时，即在生产者只负责产生并发的数据流事件，而由消费者来监听订阅并定义处理并发事件的变化传递方式，这其中要求各个步骤都是使用异步的、消息驱动的方式处理任务，这样才会节省性能。

总结来说，响应式就是一种基于 数据流和变化传递的声明式编程范式。

声明式我们在函数式编程中接触的也比较多了， java8 stream 里lamba 和各种函数的组合都可以认为是声明式的编程典型。

**背压（回压）**

在数据流的处理方式上，一般有两种方式，推和拉（这和大部分流的处理方式一样，比如 Feed 流的写扩散和读扩散），拉模式是消费者主动从生产者拉取数据，推模式是生产者主动推送数据给消费者，相较于拉模式，数据处理的资源利用率会更好，但是如果当生产者和消费者速度不一致时，那么就需要一种反馈机制达到流控的目的，背压就是这样的一种下游能够向上游反馈流量请求的一种机制。

背压是响应式编程中核心特质，一般是出现在有 buffer 上限的系统中，当出现 Buffer 溢出的时候，也就是下游消费跟不上，数据就会在传输管道中积压，就会对上游形成压力，也就是产生了 Backpressure。

### 2.3、一些响应式框架 ###

在响应式宣言中说了构建响应式系统最基本的手段是基于消息驱动来实现异步非阻塞的数据流处理，那么在非阻塞的处理模型上有大量的框架实现，如 RxJava，[Akka][]，[Vert.x][]，[Reactor][]。

其中Akka 是 scala 编写的以 [Actor 模型][Actor]（强调消息不变性，一切皆是 Actor）为基础的高性能事件驱动可伸缩并发处理框架，Actor是一个封装了状态和行为的对象，Actor之间可以通过交换消息的方式进行通信。

Vert.x 也是用于构建响应式应用的全栈式web 框架，采用事件驱动的异步工作模式，在典型的分布式领域如微服务中，性能可能是最好的，但是貌似国内好像用的不多，我也没用过，感觉不太活跃。

而 Reactor 和 RxJava 是实现了响应式规范最出名的两个框架，具体来说是 基于 Reactive Stream API 实现的。

**Reactor**

据官网介绍Reactor 是基于 Reactive Stream 的第四代响应式编程库，而SpringWebflux也是基于 Reactor 实现的，背靠 Spring，在 reactor 中核心的类有两个 Flux 和 Mono，分别代表多个数据和单个数据。

![format_png 7][]

**RxJava**

RxJava 是ReactiveX项目中的Java实现，早于Reactive Streams规范，但是使用的术语有所不同，之前是在 android 里比较火，方便了 IO 线程和 UI 线程来回切换，从 api 来看，这就是扩展的观察者模式，在 rxjava 中主要的类是 Completable，MayBe，Single，Observable，Flowable。

![format_png 8][]

**SpringWebflux**

基于 Reactor3 构建，虽然官网上说了 springweblufx 的目标一般不是为了加快应用，而是为了解决在高并发场景下使用更少的线程更少的资源来达到更低的 cpu load，占用更小的内存，获得更好的弹性负载。总结来说，解决了在高并发下IO 密集场景下的线程资源问题。

官网给的对于 springmvc 和 springwebflux 的建议中，其中有一点是要用户更多从技术原理（非阻塞 IO、并发性能、吞吐量）来评估 WebFlux 以及在团队内学习使用的成本来评估，可以看出不能无脑用，收益也需要根据实际评估。

下面看个例子，一个结合 server-sent的时钟 demo。

![format_png 9][]

顺便提下，RSocket 的[java 实现][java]也是基于 reactor 支持了响应式。（RSocket 是一个构建在 tcp 上的支持多路以及全双工通信的应用层协议，有机会不妨用 roscket-java 替代 http 体验下）

之前我们都是行走在 servlet 的世界，现在开始拥抱reactive 的世界把。(boot3也发布了，主要支持AOT 和GraalVM，基于 spring6，虚拟线程也来了)

![format_png 10][]

应用层有 webflux 的支持，那么其他比如 dao，cache 层也有响应式的框架实现了。

比如 spring-data 系列中的[spring-boot-starter-data-redis-reactive][]、[spring-boot-starter-data-mongodb-reactive][] 等

最为关注的 jdbc 也有对应的支持了，[r2dbc.io/][r2dbc.io] （Reactive 本来不支持 JDBC，最根本的原因是，JDBC 不是 non-blocking 设计）

这里粘下官网例子：

![format_png 11][]

在JDK9 中初步提供了响应式编程的规范实现： (java.util.concurrent.Flow)\[[docs.oracle.com/javase/9/do…][docs.oracle.com_javase_9_do]\]

JDK9中的 Flow 接口，从下图可以看到 只有一个 Flow 接口，在 concurrent 包下，其他四个类都是内部类，使用非常方便，可以在 jdk http 包下看到很多内部应用。

![format_png 12][]

**总结：**

在项目中替换旧有的组件并集成这些响应式的组件并不困难，问题是即使你替换了那你的应用很可能也不是响应式的系统。我之前替换过这些组件，且不说替换之后的很多隐藏的问题，就查询处理性能效果来说，并没有发生多大的改变，后来我才明白，就像刚开始提的，一个响应式的系统要求系统中各个部分都需要响应式的提供服务，而不是某个组件，一条响应式链路中某个节点如果只能同步阻塞，那么这条链路他就不是响应式的，由此带来的问题就是改造一个现有的系统为响应式，那将是一个巨大的灾难。使用反应式编程模型并不自动意味着反应式系统或反应式架构，单对于构建异步非阻塞的程序将是非常有利的。

响应式编程的好处就是执行的代码和执行的线程是分开的，通过这种方式来进行异步编程，但是因为响应式编程以及调试复杂度很高，也在不断优化寻求新的线程模型，比如协程，project loom就是在这样的背景下产生的一种新的线程并发模型。异步，不只有线程、future 和回调。

响应式的世界里，离不开异步非阻塞的讨论，这里的异步指的是什么？非阻塞呢？有必要弄清楚。

[Reactive Manifesto]: https://link.juejin.cn/?target=https%3A%2F%2Fwww.reactivemanifesto.org%2F
[format_png]: https://img-blog.csdnimg.cn/img_convert/2b2fe275a1613f18ff3f50dfe125a65e.webp?x-oss-process=image/format,png
[reactive-streams-jvm]: https://link.juejin.cn/?target=https%3A%2F%2Fkgithub.com%2Freactive-streams%2Freactive-streams-jvm
[format_png 1]: https://img-blog.csdnimg.cn/img_convert/e87763a81a62964593761f1979d4bf31.webp?x-oss-process=image/format,png
[format_png 2]: https://img-blog.csdnimg.cn/img_convert/9e16e23666d889541cb18a1ae1e741e8.webp?x-oss-process=image/format,png
[format_png 3]: https://img-blog.csdnimg.cn/img_convert/853f56e35126c02a2e2dd3b7a0392636.webp?x-oss-process=image/format,png
[format_png 4]: https://img-blog.csdnimg.cn/img_convert/06f0eb82d9cb00d5f03cc98f73328966.webp?x-oss-process=image/format,png
[format_png 5]: https://img-blog.csdnimg.cn/img_convert/8b66301e8fc7143f7bc8bed3a6ca5266.webp?x-oss-process=image/format,png
[format_png 6]: https://img-blog.csdnimg.cn/img_convert/0d14c67a448604d090633b6ced6513bc.webp?x-oss-process=image/format,png
[Akka]: https://link.juejin.cn/?target=https%3A%2F%2Fdoc.akka.io%2Fdocs%2Fakka%2Fcurrent%2Fstream%2Findex.html
[Vert.x]: https://link.juejin.cn/?target=https%3A%2F%2Fvertx.io%2F
[Reactor]: https://link.juejin.cn/?target=https%3A%2F%2Fprojectreactor.io%2F
[Actor]: https://link.juejin.cn/?target=https%3A%2F%2Fen.wikipedia.org%2Fwiki%2FActor_model
[format_png 7]: https://img-blog.csdnimg.cn/img_convert/a9b378b661d2af9196848b57be4bd0bc.webp?x-oss-process=image/format,png
[format_png 8]: https://img-blog.csdnimg.cn/img_convert/b90e1f04942910b0c030f6498e0f3d58.webp?x-oss-process=image/format,png
[format_png 9]: https://img-blog.csdnimg.cn/img_convert/138ab14a437a873b39080184c8061e63.webp?x-oss-process=image/format,png
[java]: https://link.juejin.cn/?target=https%3A%2F%2Fgithub.com%2Frsocket%2Frsocket-java
[format_png 10]: https://img-blog.csdnimg.cn/img_convert/74948ca2e5f9f7c106e6d76161ec78ed.webp?x-oss-process=image/format,png
[spring-boot-starter-data-redis-reactive]: https://link.juejin.cn/?target=https%3A%2F%2Fmvnrepository.com%2Fartifact%2Forg.springframework.boot%2Fspring-boot-starter-data-redis-reactive
[spring-boot-starter-data-mongodb-reactive]: https://link.juejin.cn/?target=https%3A%2F%2Fmvnrepository.com%2Fartifact%2Forg.springframework.boot%2Fspring-boot-starter-data-mongodb-reactive
[r2dbc.io]: https://link.juejin.cn/?target=https%3A%2F%2Fr2dbc.io%2F
[format_png 11]: https://img-blog.csdnimg.cn/img_convert/a424ca0d8401f40d7c92ef600e2105ee.webp?x-oss-process=image/format,png
[docs.oracle.com_javase_9_do]: https://link.juejin.cn/?target=https%3A%2F%2Fdocs.oracle.com%2Fjavase%2F9%2Fdocs%2Fapi%2Fjava%2Futil%2Fconcurrent%2FFlow.html
[format_png 12]: https://img-blog.csdnimg.cn/img_convert/e83c20863216332b0cfd1b6c27e9cd13.webp?x-oss-process=image/format,png