响应式编程系列一《规约》

梦里梦外; 2022-05-05 20:30 293阅读 0赞

提升开发效率，降低维护成本一直是开发团队永恒不变的宗旨。近两年来国内的技术圈子中越来越多的开始提及ReactiveX，越来越多的应用和面试中都会有ReactiveX，**响应式编程**中RxJava可谓如鱼得水。

**目录**

1. 背景

2. 响应式编程是什么

2.1 原理简析

2.3 与传统观察者模式不同

2.2 Rx是Push还是Pull

3. 优势 & 代价

4. Reactive Streams规约

4.1 Publisher

4.2 Subscriber

4.3 Subscription

4.4 Processor

5. 主流实现

5.1 Rx2.x

5.2 Reactive Stream

5.3 Java9

5.4 Spring WebFlux

5.5 Vertx

--------------------

## 1. 背景 ##

前面提到过是为了提升开发效率，代码质量（**简洁**），降低维护成本。

注：异步操作很关键的一点是程序的**简洁性**，因为在调度过程比较复杂的情况下，异步代码经常会既难写也难被读懂。

## 2. 响应式编程是什么 ##

响应式编程是一种**基于异步数据流概念的编程模式**。数据流就像一条河：它可以被观测，被过滤，被操作，或者为新的消费者与另外一条流合并为一条新的流。响应式编程的一个关键概念是**事件**。事件可以被等待，可以触发过程，也可以触发其它事件。

Rx提供了一系列的操作符，你可以使用它们来过滤(filter)、选择(select)、变换(transform)、结合(combine)和组合(compose)多个Observable，这些操作符让执行和复合变得非常高效。

响应式流从2013年开始，作为提供**非阻塞背压**的**异步流处理**标准的倡议。它旨在解决处理元素流的问题——如何将元素流从发布者传递到订阅者，而**不需要发布者阻塞**，或订阅者有无限制的缓冲区或丢弃。

在2015年，出版了用于处理响应式流的规范和Java API。请访问[http://www.reactive-streams.org/][http_www.reactive-streams.org] 。

### 2.1 原理简析 ###

Rx 的异步实现，是通过一种**扩展的观察者模式**来实现的。程序的观察者模式和这种真正的『观察』略有不同，观察者不需要时刻盯着被观察者（例如 A 不需要每过 2ms 就检查一次 B 的状态），而是采用**注册**(Register)**或者称为**订阅**(Subscribe)**的方式，告诉被观察者：我需要你的某某状态，你要在它变化的时候通知我。

![70][]

![20190304135021498.png][]

如上图可以引出Rx的一些重要概念：

1.  Button（被观察者）作为事件的生产方（产生“onClick”事件），是主动生产的，是整个事件流程的起点；
2.  OnClickListener（观察者）作为事件的处理方（处理“onClick”事件）是被动的，是整个事件流程的终点；
3.  在起点跟终点之间，即事件传递的过程中是可以被加工，过滤，转换，合并等等方式处理，这就是Rx中常说的操作符的职能

![20190304135105159.png][]

Observable（被观察者）可以理解为**数据源**、**数据发射器**，Observer（观察者）针对行为事件数据做出的响应。`Observable` 和 `Observer` 通过 `subscribe()` 方法实现订阅关系，从而 `Observable` 可以在需要的时候发出事件来通知 `Observer`。

### 2.3 与传统观察者模式不同  ###

RxJava 的事件回调方法除了普通事件 onNext （相当于 onClick / onEvent）之外，还定义了两个特殊的事件：onCompleted 和 onError。

![20190304140926842.png][]

**为何需要会多两个onCompleted/onError呢？**

Observable为了更贴近有Iterable类型操作一致，添加了两种缺少的语义：

*  生产者可以发信号给消费者，通知它没有更多数据可用了（对于Iterable，一个for循环正常完成表示没有数据了；对于Observable，就是调用观察者的onCompleted方法）；
 *  生产者可以发信号给消费者，通知它遇到了一个错误（对于Iterable，迭代过程中发生错误会抛出异常；对于Observable，就是调用观察者(Observer)的onError方法）

### 2.2 Rx是Push还是Pull ###

响应式流模型非常简单——订阅者向发布者发送多个元素的异步请求。 发布者向订阅者异步发送多个或稍少的元素。响应式流在pull模型和push模型流处理机制之间动态切换。 当订阅者较慢时，它使用pull模型，当订阅者更快时使用push模型。

你可以把Observable当做Iterable的推送方式的等价物，可以理解为：**所有使用Iterable的地方都可以使用Rx**（上面也提过），

*  使用Iterable，消费者从生产者那**拉取**数据，线程**阻塞**直至数据准备好；
 *  使用Observable，在数据准备好时，生产者将数据**推送**给消费者（**可同步可异步**更灵活）

<table> 
 <tbody> 
 <tr> 
 <td> 事件 </td> 
 <td> Iterable(pull) </td> 
 <td> Observable(push) </td> 
 </tr> 
 <tr> 
 <td> 获取数据 </td> 
 <td> <code>T next()</code> </td> 
 <td> <code>onNext(T)</code> </td> 
 </tr> 
 <tr> 
 <td> 异常处理 </td> 
 <td> throws <code>Exception</code> </td> 
 <td> <code>onError(Exception)</code> </td> 
 </tr> 
 <tr> 
 <td> 任务完成 </td> 
 <td> <code>!hasNext()</code> </td> 
 <td> <code>onCompleted()</code> </td> 
 </tr> 
 </tbody> 
</table>

//伪代码：
    // Iterable
    getDataFromLocalMemory()
      .skip(10)
      .take(5)
      .map({ s -> return s + " transformed" })
      .forEach({ println "next => " + it });
    
    // Observable
    getDataFromNetwork()
      .skip(10)
      .take(5)
      .map({ s -> return s + " transformed" })
      .subscribe({ println "onNext => " + it });

总之，Observable是**异步**的**双向**push，Iterable是**同步**的**单向**pull。

## 3. 优势 & 代价 ##

**优势**

*  函数式风格：对可观察数据流使用无副作用的输入输出函数，避免了程序里错综复杂的状态；
 *  简化代码：Rx的操作符通通常可以将复杂的难题简化为很少的几行代码（声明式的组合这些序列）；
 *  异步错误处理：传统的try/catch没办法处理异步计算，Rx提供了合适的错误处理机制；
 *  轻松使用并发：Rx的Observables和Schedulers让开发者可以摆脱底层的线程同步和各种并发问题；

**代价**

1.  虽然复用线程有助于提高吞吐量，但一旦在某个回调函数中线程被卡住，那么这个线程上所有的请求都会被阻塞，最严重的情况，整个应用会被拖垮。
2.  难以调试。由于 Rx 强大的描述能力，在一个典型的 Rx 应用中，大部分代码都是以链式表达式的形式出现，比如flux.map(String::toUpperCase).doOnNext(s -> LOG.info("UC String \{\}", s)).next().subscribe()，一旦出错，你将很难定位到具体是哪个环节出了问题。所幸的是，Rx 框架一般都会提供一些工具方法来辅助进行调试。

## 4. Reactive Streams规约 ##

### 4.1 Publisher ###

是潜在无限数量的有序元素的生产者。 它根据收到的要求向当前订阅者发布（或发送）元素

### 4.2 Subscriber ###

从发布者那里订阅并接收元素。 发布者向订阅者发送订阅令牌（subscription token）。 使用订阅令牌，订阅者从发布者哪里请求多个元素。 当元素准备就绪时，发布者向订阅者发送多个或更少的元素。 订阅者可以请求更多的元素。 发布者可能有多个来自订阅者的元素待处理请求。

### 4.3 Subscription ###

表示订阅者订阅的一个发布者的令牌。 当订阅请求成功时，发布者将其传递给订阅者。 订阅者使用订阅令牌与发布者进行交互，例如请求更多的元素或取消订阅。

![70 1][]

### 4.4 Processor ###

充当订阅者和发布者的处理阶段。`Processor<T,R>`订阅类型T的数据元素，接收并转换为类型R的数据，并发布变换后的数据，该接口继承了`Publisher`和`Subscriber`接口。

![70 2][]

## 5. 主流实现 ##

### 5.1 Rx2.x ###

[RxJava][]是响应式流的Java实现之一，而RxJava 2.0 已经按照Reactive-Streams specification规范完全的重写了。

### 5.2 Reactive Stream ###

[Reactor 2.0.0.RC1][] 于2015年02月19日由Pivotal RTI（Spring 框架发起者）发布，支持 Reactive Stream，它的构架总览：

![70 3][]

注：上图参考附录Reactor指南中文版，Reactor1.x实在是不太出名，也是规范没出来吧

Reactor 代码库拆分成多个子模块，便于选择所需功能，不受其他功能代码块干扰。

下面举例说明，为实现异步目标，响应式技术和 Reactor 模块该如何搭配：

1.  Spring XD + Reactor-Net (Core/Stream)： 使用 Reactor 作为 Sink/Source IO 驱动。
2.  Grails | Spring + Reactor-Stream (Core)： 用 Stream 和 Promise 做后台处理。
3.  Spring Data + Reactor-Bus (Core)： 发射数据库事件 (保存/删除/…)。
4.  Spring Integration Java DSL + Reactor Stream (Core)： Spring 集成的微批量信息通道。
5.  RxJavaReactiveStreams + RxJava + Reactor-Core： 融合富结构与高效异步 IO 处理
6.  RxJavaReactiveStreams + RxJava + Reactor-Net (Core/Stream)： 用 RxJava 做数据输入，异步 IO 驱动做传输。

**与Rx2.x****的差异**：

<table> 
 <tbody> 
 <tr> 
 <td style="border-color:#a5a5a5;width:91.9pt;"> RxJava </td> 
 <td style="width:195px;"> reactor-stream </td> 
 <td style="border-color:#a5a5a5;width:403px;"> 说明 </td> 
 </tr> 
 <tr> 
 <td style="border-color:#c9c9c9;width:91.9pt;"> Observable </td> 
 <td style="width:195px;"> reactor.rx.Stream </td> 
 <td style="width:403px;"> Reactive Stream Publisher的实现 </td> 
 </tr> 
 <tr> 
 <td style="border-color:#c9c9c9;vertical-align:top;width:91.9pt;"> Operator </td> 
 <td style="vertical-align:top;width:195px;"> reactor.rx.action.Action </td> 
 <td style="vertical-align:top;width:403px;"> Reactive Stream Processor的实现 </td> 
 </tr> 
 <tr> 
 <td style="border-color:#c9c9c9;width:91.9pt;"> Observable with 1 data at most </td> 
 <td style="width:195px;"> reactor.rx.Promise </td> 
 <td style="width:403px;"> 返回唯一结果的类型, &nbsp;Reactive Stream Processor实现并提供了可选的异步分发功能。&nbsp; </td> 
 </tr> 
 <tr> 
 <td style="border-color:#c9c9c9;vertical-align:top;width:91.9pt;"> Factory API (just, from…) </td> 
 <td style="vertical-align:top;width:195px;"> reactor.rx.Streams </td> 
 <td style="vertical-align:top;width:403px;"> 和core模块的 data-focused 子类一样, 返回 Stream </td> 
 </tr> 
 <tr> 
 <td style="border-color:#c9c9c9;width:91.9pt;"> Functional API (map, filter…) </td> 
 <td style="width:195px;"> reactor.rx.Stream </td> 
 <td style="width:403px;"> 和core模块的data-focused 子类一样, 返回Stream </td> 
 </tr> 
 <tr> 
 <td style="border-color:#c9c9c9;vertical-align:top;width:91.9pt;"> Schedulers </td> 
 <td style="vertical-align:top;width:195px;"> reactor.core.Dispatcher, org.reactivestreams.Processor </td> 
 <td style="vertical-align:top;width:403px;"> Reactor Stream计算无限制的共享Dispatcher或者有限的Processor的操作。 </td> 
 </tr> 
 <tr> 
 <td style="border-color:#c9c9c9;width:91.9pt;"> Observable.observeOn() </td> 
 <td style="width:195px;"> Stream.dispatchOn() </td> 
 <td style="width:403px;"> 只是dispatcher参数的一个适配命名。 </td> 
 </tr> 
 </tbody> 
</table>

### 5.3 Java9 ###

JDK 9 java.util.concurrent 包提供了两个主要的 API 来处理响应流：

1.  Flow
2.  SubmissionPublisher

### 5.4 Spring WebFlux ###

Spring WebFlux 是 Spring 5 的一个新模块，包含了响应式 HTTP 和 WebSocket 的支持，在容器中 Spring WebFlux 会将输入流适配成 Mono 或者 Flux 格式进行统一处理。，另外在上层服务端支持两种不同的编程模型：  
\- 基于 Spring MVC 注解 @Controller 等  
\- 基于 Functional 函数式路由

![70 4][]

**为啥只能运行在 Servlet 3.1+** **容器？**3.1 规范其中一个新特性是异步处理支持。

### 5.5 Vertx ###

[Vert.x][]是一个异步无阻塞的网络框架，其参照物是node.js。基本上node.js能干的事情，Vert.x都能干。Vert.x利用Netty4的EventLoop来做单线程的事件循环，所以跑在Vert.x上的业务不能做CPU密集型的运算，这样会导致整个线程被阻塞。

Vert.x目前是见过**功能最强大（****core****、web****、Data access****、reacive****、microservices****、MQTT****）**，第三方库依赖最少的Java框架，它只依赖Netty4以及Jacskon，另外如果你需要建立分布式的Vert.x则再依赖HazelCast这个分布式框架，注意Vert.x3必须基于Java8。

总结，响应式编程已经慢慢成我我们开发中的主流，后续将带您深入了解《[rxjava][]》![89576152][]

[http_www.reactive-streams.org]: http://www.reactive-streams.org/
[70]: /images/20220506/7f9d38bf3c264cafa9e9453f1395d6dd.png
[20190304135021498.png]: /images/20220506/606f1f9a9bcd4d269227669abfdc0e77.png
[20190304135105159.png]: /images/20220506/a74be90216bd41c6ab587286ab01ab80.png
[20190304140926842.png]: /images/20220506/22fddc9229a24dceb5458626628d3033.png
[70 1]: /images/20220506/9ef93ced433747fba5b4578902e25fe4.png
[70 2]: /images/20220506/6ceabd83d18b4fa7b3ca2fc329fe8238.png
[RxJava]: https://github.com/ReactiveX/RxJava
[Reactor 2.0.0.RC1]: https://github.com/reactor/reactor-core
[70 3]: /images/20220506/1f948f1edab2456fbb8af5d8c4c085a3.png
[70 4]: /images/20220506/93ebb1179381436bb0a51a16a506a05e.png
[Vert.x]: https://vertx.io/docs/
[rxjava]: https://blog.csdn.net/alex_xfboy/article/details/83089734
[89576152]: /images/20220506/23b36bfea1f849ea9ebcbd0a2b93c59d.png