Java：理解java响应式编程

怼烎@ 2023-01-06 11:59 135阅读 0赞

原文：Understanding reactive programming in Java  
[https://nullbeans.com/understanding-reactive-programming-in-java/][https_nullbeans.com_understanding-reactive-programming-in-java]

这篇文章我们将讨论响应式编程的原理，它要解决的问题，以及java响应式编程的基础。这篇教程聚集于java响应式编程的用法，但其中讨论的原理和思路同样适用于其它编程语言。

这篇文章可能读起来很长，但如果你对响应式编程有疑问，并且不知道从何处开始，或者对于java响应式编程难于找到好的信息来源，那么这篇文章会是一个好的开始并能帮你澄清很多主题。

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

**目录**

响应式编程要解决什么问题？

响应式程序的执行

响应式编程是什么？

响应式声明

什么是背压（Backpressure）

Java响应式编程

发布者（Publisher）

订阅者（Subscriber）

订阅（Subscription）

处理器（Processor）

综述

总结

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

## 响应式编程要解决什么问题？ ##

当我开始研究响应式编程时，我可能会读到像这样的定义：响应式编程是关注于数据交换和传播的编程范式。然而，这一定义丢掉了这一编程范式的核心且并没有更多的解决。

那么，我们试着把事情简化一下，只强调传统命令式程序执行和异步“响应”执行之间的不同之处。

一个程序由一系列计算步骤组合而成，它们需要按特定的顺序执行。传统程序执行时，线程一次只执行一个步骤。我们以快餐店为例。我们假设在收银台只有一个服务员。这个收银员负责接单，然后厨师将食物做好后打包食物。

每个顾客会直到柜台点餐。服务员将订单传给厨师，在食物准备好前会等待几分钟，然后打包并递给顾客。然后，服务员开始接下一个顾客下的订单。

这种方式的问题在于，由于浪费了空闲时间所以没有效率。首先，服务员等待厨师制作食物时时间就已浪费了。在此期间，除了等待没做任何事情。同时，厨师当服务员接下一个顾客的订单时也处于空闲状态。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25ldHllYXhp_size_16_color_FFFFFF_t_70][]

程序同步执行的例子

在这个例子中，服务员代表一个线程，而厨师代表一个数据库，顾客的订单代表用户输入。在传统模式中程序执行时，比如，获取一个REST请求，线程将首先将请求转化为数据库查询，然后发送到数据库，等待它的返回，一旦获取返回，线程处理完成后就会给顾客一个响应。

在线程等待数据库返回时，线程处于阻塞状态。这将导致程序执行无效率而且还慢。尽管可以增加线程池中可用线程数，但每个线程自身也会有内存开销，最终还是会受到处理器核心数的限制。

同时，如果应用依赖于i/o系统，比如网络资源或数据库，添加更多的线程不会提高由i/o时廷造成的性能问题。

**    总之，响应式编程的目是帮助应用在不同的环境条件和负载下仍保持响应性。**

## 响应式程序的执行 ##

响应式执行结合了一些异步执行的原则，并以一种特定模式运行，这种模式我们可以通过“响应式编程”来定义并控制。

我们回顾一下快餐店的例子。在响应式执行场景中，服务员（线程）从顾客那是接单（用户输入），然后传给厨师（数据库）。服务员非但没有等待厨师完成订单，他告诉顾客（订阅者）一旦订单完成会通知他们（通过订阅）。服务员会继续从下一个顾客那里接单并执行相同的步骤。

服务员会一直接顾客的订单直到厨师通知服务员一笔订单已做完。当订单已完成，服务员会递给顾，客顾客的请求也随之完成。

这种方法以两种方式提高了效率。首先，服务员没以空闲等待。接着，厨师可以收到稳定的订单借给去制作食物，使得厨师的资源更充分的被利用。这样也会提高顾客的体验，因为他们会更快地被接待且订单会被处理的更快，而不会排长队等待得不到接待。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25ldHllYXhp_size_16_color_FFFFFF_t_70 1][]

异步响应式执行。服务员接单传给厨师，然后再接更多的订单，这是一种执行然后忘记模式

这就是传统同步执行与异步响应式执行之间最大的不同。**也就是说，响应式编程能在线程级别提高执行效率。**

## 响应式编程是什么？ ##

现在我们知道响应式程序是怎么执行的，也就是说，**响应式编程是一种编程范式，基于它的API、库和语言特性会采用特定的设计模式，这种设计模式的目的是完成异步响应式程序的执行。**我们将在下面的章节探索响应式设计模式。

## 响应式声明 ##

应用如果要变成响应式的，它需要拥有特定的属性。这些属性被定义在现在称为“响应式声明”中，它使得“响应式的”就用不同于异步应用。

应用如果要变成响应式的，它需要具有：

> **可应答的（Responsive）**：系统需要及时对请求进行应答。可应答的系统应该具有快速和一致的响应时间。
> 
> **弹性的（Resilient）：**一旦出现故障，系统应仍可提供应答。例如，用户尝试访问一个出错的网站，网站应给用户回复一个好的错误页面而不是什么也不回复。
> 
> **可伸缩的（Elastic）**：一个可伸缩的系统可适应不同的负载。例如，负载达到峰值时可对资源扩容，负载变低时可释放资源。可伸缩的系统在高负载时仍应是可应答的。
> 
> **消息驱动的（Message Driven）**：响应式系统利用异步消息驱动的通讯机制（服务员告诉顾客一旦订单完成就会通知）。这样资源使用效率更高，故障隔离性和容忍性会更好。想像一下你的系统正在使用可能会随时挂掉的不稳定的数据库。如何使用传统阻塞调用，每次会有一个应用线程因调用数据库而挂起，这个线程会被阻塞很长时间。这会使应用中剩下的可用线程变少。异步消息驱动的通讯方式可以解决这种问题。

注意这4 个原则仅仅只是原则。没有系统可以免于故障或性能降级。然而，你的目标是建设可响应的系统，所以要尽可能的遵从这些原则。

响应式声明是对可响应的系统一般性理解，它由技术工业领域里的一些个人或组织总结的。你可以在[这里][Link 1]看到更多关于它的内容。

## 什么是背压（Backpressure） ##

背压是另外一个关于数据流速控制复杂名词。它是为了提高系统的稳定性和完整性才被使用的。

拿一个处理用户查询的应用来举例。系统会处理用户发送来的每一个请求。然而，如果一个用户请求发的又多又快，他们会使系统变慢或使它挂掉，并降低其他用户的体验（比如服务拒绝攻击）。

背压试图通过限制在指定时间内允许的操作数来解决这个问题（也就是限制执行速率）。

响应式编程可以采用多种策略实现背压，比如使用缓存或拒绝请求。我们会在另一篇文章中探讨这些策略。

## Java响应式编程 ##

如果你搜索Java响应式编程，你很可能会被网上不同的实现和不同教程和文章中的代码所困惑。这是有原因的。

当前，Java中没有标准统一的响应式API实现。当前有大量的库提供不同的实现，以及工具运行响应式编程。

从RxJava 1 和2，Java SDK 9中引入的Flow API ，Reactive Streams，到Project Reactor (Spring在使用的)和 Akka Streams，这里只列举其中的一些。

需要记住的重要事情是这些框架都基于相同的设计模式，它们中的每一种都或多或少的提供开箱即用的功能和方便使用的类。我们将在后面的教程中探索这些不同的库。

我们将聚焦于大多数框架基于的基本设计模式。最终要选择哪个库由你自己来定。

我们以定义响应式程序的基本组件开始，在最后我们会总结一下它们一起是怎么工作的。

### 发布者（**Publisher**） ###

发布者就是数据生产者。这个组件为系统产生想要的数据。在实践中，这可能是一个数据库的查询结果、twitter feed、股票市场报价feed，等等。

发布者没有名字...或者说许多名字。它依赖于响应式库的选择。

>  *  **Observable** (RxJava) / **Mono** (Project Reactor)： 这种**不产生任何数据**的发布者有许多名字。在Reactive Streams的规范中，它就叫做发布者。
>  *  **Single** (RxJava) / **Mono** (Project Reactor)： 这种**最多产生一条数据**的发布者有许多名字。在Reactive Streams的规范中，它就叫做发布者。
>  *  **Observable** (RxJava) / **Flux** (Project Reactor)： 这种**产生多条数据**的发布者有许多名字。在Reactive Streams的规范中，它仍就叫做发布者。

### 订阅者（**Subscriber**） ###

订阅者“订阅”发布者，当新数据产生、错误发生、或发布者完成数据生产后会收到通知。产生的数据会在订阅者中进行处理。

幸运的是，订阅者在RxJava和Project Reactor中还是被叫做订阅者。

订阅者有3个数据通道。每一个通道都是通过订阅者的内部方法进行通知。

>  *  **OnNext**：发布者产生一条或多条数据项时会调用此方法。
>  *  **OnError**：发布者在生成数据过程中产生错误时会调用此方法。
>  *  **OnComplete**：此方法被调用说明发布者数据生产已完成，不会再数据产生了。

请注意以上方法可以在调用方线程或单独的线程被调用，这依赖于你的配置。

最后要讨论的是**OnSubscribe**方法。这个方法只会在订阅创建时被调用一次。注意它不被看作是数据通道。它被用来设置订阅者以及从订阅中请求第一条数据。

### 订阅（**Subscription**） ###

订阅是一个订阅者和发布者之间的合约。订阅者使用它从发布者那里请求更多的数据项，或者取消订阅中止接收事多的数据。

订阅者能通过订阅中的**请求**方法从发布者那里请求后续1~n条数据。这个通常在订阅者的onSubscribe和onNext方法中完成。

### 处理器（**Processor**） ###

处理器是一种即可扮演订阅者又可扮演发布者的响应式实体。它即能消费由发布者产生的数据项，也可以自己发布数据。

处理器通常在发布者和消费者之间安装，用来处理中间数据项。比如，你可以创建处理器对发布者产生的所的单词执行拼写检查。这种情况下，处理器作为订阅者。处理器发布更正过的单词给它的订阅者们，这时它又成为了发布者。

### 综述 ###

一张图胜过千言万语。所以我看一下下面的图：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25ldHllYXhp_size_16_color_FFFFFF_t_70 2][]

响应式程序的执行例子

在一般的响应式程序执行过程中，总是以订阅者订阅发布者开始。一旦发布者与订阅者之间的订阅成功创建，订阅者会使用订阅中的请求方法请求数据。

当发布者发布新数据时，订阅者的onNext方法被调用。订阅者能请求更多数据或取消订阅。在我们的例子中，订阅者请求更多数据。

一旦发布者没有数据楞发布了，订阅者的onComplete方法会被调用，说明合约（订阅）完成并结束。

## 总结 ##

响应式编程已经有一段时间了。然而，技术行业最终开始关注这一设计模式是由于对更好地响应时间的要求、更严格地可靠性要求以及微服务架构的兴起。

在i/o场景中为了更好的利用可用系统资源以及提高应用响应时间，我们在这个教程中讨论了使用响应式编程的优点。

[https_nullbeans.com_understanding-reactive-programming-in-java]: https://nullbeans.com/understanding-reactive-programming-in-java/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25ldHllYXhp_size_16_color_FFFFFF_t_70]: /images/20221119/96402bf4822549dc85ebe32277fcce42.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25ldHllYXhp_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20210114184546247.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25ldHllYXhp,size_16,color_FFFFFF,t_70
[Link 1]: https://www.reactivemanifesto.org/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25ldHllYXhp_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20210115135735206.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25ldHllYXhp,size_16,color_FFFFFF,t_70