快捷導(dǎo)航

Spring5新特性之Reactive響應(yīng)式編程

更新時(shí)間：2022年03月06日 14:59:53 作者：紅帽海綿寶寶

這篇文章主要介紹了Spring5新特性之Reactive響應(yīng)式編程，響應(yīng)式編程是一種編程范式，通用和專注于數(shù)據(jù)流和變化的，并且是異步的，下文更多詳細(xì)內(nèi)容，需要的小伙伴可以參考一下，希望對(duì)你有所幫助

1 什么是響應(yīng)式編程

一句話總結(jié)：響應(yīng)式編程是一種編程范式，通用和專注于數(shù)據(jù)流和變化的，并且是異步的。

維基百科原文：

In computing, reactive programming is an asynchronous programming paradigm concerned with data streams and the propagation of change. This means that it becomes possible to express static (e.g. arrays) or dynamic (e.g. event emitters) data streams with ease via the employed programming language(s), and that an inferred dependency within the associated execution model exists, which facilitates the automatic propagation of the change involved with data flow.

翻譯：

在計(jì)算機(jī)領(lǐng)域，響應(yīng)式編程是一個(gè)專注于數(shù)據(jù)流和變化傳遞的**異步編程范式。**這意味著可以使用編程語(yǔ)言很容易地表示靜態(tài)（例如數(shù)組）或動(dòng)態(tài)（例如事件發(fā)射器）數(shù)據(jù)流，并且在關(guān)聯(lián)的執(zhí)行模型中，存在著可推斷的依賴關(guān)系，這個(gè)關(guān)系的存在有利于自動(dòng)傳播與數(shù)據(jù)流有關(guān)的更改。

舉例：

例如，在命令式編程環(huán)境中， a:=b+c 表示將表達(dá)式的結(jié)果賦給a ，而之后改變b 或c 的值不會(huì)影響。但在響應(yīng)式編程中，a的值會(huì)隨著b或c 的更新而更新。電子表格程序就是響應(yīng)式編程的一個(gè)例子。單元格可以包含字面值或類似"=B1+C1"的公式，而包含公式的單元格的值會(huì)依據(jù)其他單元格的值的變化而變化。

響應(yīng)式編程最初是為了簡(jiǎn)化交互式用戶界面的創(chuàng)建和實(shí)時(shí)系統(tǒng)動(dòng)畫的繪制而提出來的一種方法，但它本質(zhì)上是一種通用的編程范式。

2 回顧Reactor

2.1 什么是Reactor

還是維基百科：

The reactor design pattern is an event handling pattern for handling service requests delivered concurrently to a service handler by one or more inputs. The service handler then demultiplexes the incoming requests and dispatches them synchronously to the associated request handlers.

翻譯：

反應(yīng)器（reactor）設(shè)計(jì)模式是一種事件處理模式，用于處理由一個(gè)或多個(gè)輸入并發(fā)交付給服務(wù)處理程序的服務(wù)請(qǐng)求。然后，服務(wù)處理程序?qū)魅氲恼?qǐng)求解復(fù)用，并將它們同步地分派給相關(guān)的請(qǐng)求處理程序。

2.2 為什么是Reactor

Reactor來源于網(wǎng)絡(luò)IO中同步非阻塞的I/O多路復(fù)用機(jī)制的模式。

堵塞、非堵塞的區(qū)別是在于第一階段，即數(shù)據(jù)準(zhǔn)備階段。無(wú)論是堵塞還是非堵塞，都是用應(yīng)用主動(dòng)找內(nèi)核要數(shù)據(jù)，而read數(shù)據(jù)的過程是‘堵塞’的，直到數(shù)據(jù)讀取完。
同步、異步的區(qū)別在于第二階段，若由請(qǐng)求者主動(dòng)的去獲取數(shù)據(jù)，則為同步操作，需要說明的是：read/write操作也是‘堵塞’的，直到數(shù)據(jù)讀取完。

若數(shù)據(jù)的read都由kernel內(nèi)核完成了(在內(nèi)核read數(shù)據(jù)的過程中，應(yīng)用進(jìn)程依舊可以執(zhí)行其他的任務(wù))，這就是異步操作。

換句話說，

BIO里用戶最關(guān)心“我要讀”，
NIO里用戶最關(guān)心"我可以讀了"，
在AIO模型里用戶更需要關(guān)注的是“讀完了”。

NIO一個(gè)重要的特點(diǎn)是：socket主要的讀、寫、注冊(cè)和接收函數(shù)，在等待就緒階段都是非阻塞的，真正的I/O操作是同步阻塞的（消耗CPU但性能非常高）。
NIO是一種同步非阻塞的I/O模型，也是I/O多路復(fù)用的基礎(chǔ)。

Reactor模式基本結(jié)構(gòu)：

Handle：本質(zhì)上表示一種資源(比如說文件描述符，或是針對(duì)網(wǎng)絡(luò)編程中的socket描述符)，是由操作系統(tǒng)提供的；該資源用于表示一個(gè)個(gè)的事件，事件既可以來自于外部，也可以來自于內(nèi)部，Handle是事件產(chǎn)生的發(fā)源地。
Synchronous Event Demultiplexer(同步事件分離器)：它本身是一個(gè)系統(tǒng)調(diào)用，用于等待事件的發(fā)生(事件可能是一個(gè)，也可能是多個(gè))。調(diào)用方在調(diào)用它的時(shí)候會(huì)被阻塞，一直阻塞到同步事件分離器上有事件產(chǎn)生為止。對(duì)于Linux來說，同步事件分離器指的就是常用的I/O多路復(fù)用機(jī)制，比如說select、poll、epoll等。在Java NIO領(lǐng)域中，同步事件分離器對(duì)應(yīng)的組件就是Selector；對(duì)應(yīng)的阻塞方法就是select方法。
Event Handler(事件處理器)：本身由多個(gè)回調(diào)方法構(gòu)成，這些回調(diào)方法構(gòu)成了與應(yīng)用相關(guān)的對(duì)于某個(gè)事件的反饋機(jī)制。在Java NIO領(lǐng)域中并沒有提供事件處理器機(jī)制讓我們調(diào)用或去進(jìn)行回調(diào)，是由我們自己編寫代碼完成的。Netty相比于Java NIO來說，在事件處理器這個(gè)角色上進(jìn)行了一個(gè)升級(jí)，它為我們開發(fā)者提供了大量的回調(diào)方法，供我們?cè)谔囟ㄊ录a(chǎn)生時(shí)實(shí)現(xiàn)相應(yīng)的回調(diào)方法進(jìn)行業(yè)務(wù)邏輯的處理，即，ChannelHandler。ChannelHandler中的方法對(duì)應(yīng)的都是一個(gè)個(gè)事件的回調(diào)。
Concrete Event Handler(具體事件處理器)：是事件處理器的實(shí)現(xiàn)。它本身實(shí)現(xiàn)了事件處理器所提供的各種回調(diào)方法，從而實(shí)現(xiàn)了特定于業(yè)務(wù)的邏輯。它本質(zhì)上就是我們所編寫的一個(gè)個(gè)的處理器實(shí)現(xiàn)。
Initiation Dispatcher(初始分發(fā)器)：實(shí)際上就是Reactor角色。它本身定義了一些規(guī)范，這些規(guī)范用于控制事件的調(diào)度方式，同時(shí)又提供了應(yīng)用進(jìn)行事件處理器的注冊(cè)、刪除等設(shè)施。它本身是整個(gè)事件處理器的核心所在，Initiation Dispatcher會(huì)通過Synchronous Event Demultiplexer來等待事件的發(fā)生。一旦事件發(fā)生，Initiation Dispatcher首先會(huì)分離出每一個(gè)事件，然后調(diào)用事件處理器，最后調(diào)用相關(guān)的回調(diào)方法來處理這些事件。Netty中ChannelHandler里的一個(gè)個(gè)回調(diào)方法都是由bossGroup或workGroup中的某個(gè)EventLoop來調(diào)用的。

2.3 Reactor模式的經(jīng)典實(shí)現(xiàn)—Netty

Netty的線程模式就是一個(gè)實(shí)現(xiàn)了Reactor模式的經(jīng)典模式。

結(jié)構(gòu)對(duì)應(yīng)：

NioEventLoop ———— Initiation Dispatcher
Synchronous EventDemultiplexer ———— Selector
Evnet Handler ———— ChannelHandler
ConcreteEventHandler ———— 具體的ChannelHandler的實(shí)現(xiàn)

模式對(duì)應(yīng)：

Netty服務(wù)端使用了“多Reactor線程模式”
mainReactor ———— bossGroup(NioEventLoopGroup) 中的某個(gè)NioEventLoop
subReactor ———— workerGroup(NioEventLoopGroup) 中的某個(gè)NioEventLoop
acceptor ———— ServerBootstrapAcceptor
ThreadPool ———— 用戶自定義線程池

流程：

① 當(dāng)服務(wù)器程序啟動(dòng)時(shí)，會(huì)配置ChannelPipeline，ChannelPipeline中是一個(gè)ChannelHandler鏈，所有的事件發(fā)生時(shí)都會(huì)觸發(fā)Channelhandler中的某個(gè)方法，這個(gè)事件會(huì)在ChannelPipeline中的ChannelHandler鏈里傳播。然后，從bossGroup事件循環(huán)池中獲取一個(gè)NioEventLoop來現(xiàn)實(shí)服務(wù)端程序綁定本地端口的操作，將對(duì)應(yīng)的ServerSocketChannel注冊(cè)到該NioEventLoop中的Selector上，并注冊(cè)ACCEPT事件為ServerSocketChannel所感興趣的事件。
② NioEventLoop事件循環(huán)啟動(dòng)，此時(shí)開始監(jiān)聽客戶端的連接請(qǐng)求。
③ 當(dāng)有客戶端向服務(wù)器端發(fā)起連接請(qǐng)求時(shí)，NioEventLoop的事件循環(huán)監(jiān)聽到該ACCEPT事件，Netty底層會(huì)接收這個(gè)連接，通過accept()方法得到與這個(gè)客戶端的連接(SocketChannel)，然后觸發(fā)ChannelRead事件(即，ChannelHandler中的channelRead方法會(huì)得到回調(diào))，該事件會(huì)在ChannelPipeline中的ChannelHandler鏈中執(zhí)行、傳播。
④ ServerBootstrapAcceptor的readChannel方法會(huì)該SocketChannel(客戶端的連接)注冊(cè)到workerGroup(NioEventLoopGroup) 中的某個(gè)NioEventLoop的Selector上，并注冊(cè)READ事件為SocketChannel所感興趣的事件。啟動(dòng)SocketChannel所在NioEventLoop的事件循環(huán)，接下來就可以開始客戶端和服務(wù)器端的通信了。

3 Spring5中多Reactive的支持

3.1 Spring Webflux

3.1.1 依賴

<dependency>
? ? <groupId>org.springframework.boot</groupId>
? ? <artifactId>spring-boot-starter-webflux</artifactId>
</dependency>

3.1.2 Controller代碼

@RestController
public class HelloController {

? ? @GetMapping("/hello")
? ? public Mono<String> hello() {
? ? ? ? return Mono.just("Hello Spring Webflux");
? ? }
? ??
}

3.1.3 測(cè)試

C:\Users\xxxx\Desktop\sb-reactive>curl http://localhost:8080/hello
Hello Spring Webflux

3.1.4 Spring MVC和Spring WebFlux模式上的不同

3.2 Spring Data Reactive Respositories

3.2.1 依賴

<dependency>
? ? <groupId>org.springframework.boot</groupId>
? ? <artifactId>spring-boot-starter-data-redis-reactive</artifactId>
</dependency>

3.2.2 配置

public class RedisReactiveConfig {

? ? @Bean
? ? public ReactiveRedisConnectionFactory connectionFactory() {
? ? ? ? return new LettuceConnectionFactory("127.0.0.1", 6379);
? ? }

? ? @Bean
? ? public ReactiveStringRedisTemplate reactiveStringRedisTemplate(ReactiveRedisConnectionFactory factory) {
? ? ? ? return new ReactiveStringRedisTemplate(factory);
? ? }

}

3.3.3 測(cè)試

@SpringBootTest
class SbReactiveApplicationTests {

? ? @Autowired
? ? private ReactiveStringRedisTemplate reactiveRedisTemplate;

? ? @Test
? ? void contextLoads() {
? ? ? ? reactiveRedisTemplate
? ? ? ? ? ? ? ? .opsForValue().set("1", "zs")
? ? ? ? ? ? ? ? .subscribe(b -> System.out.println("success"),
? ? ? ? ? ? ? ? ? ? ? ? e -> System.out.println("error"));
? ? }

}

4 如何理解Reactive響應(yīng)式編程？

概念有很多，但是它相較我們的一般請(qǐng)求處理到底有什么更好的價(jià)值體現(xiàn)?

解釋：

Reactive Programming 作為觀察者模式（Observer）的延伸，不同于傳統(tǒng)的命令編程方式（ Imperative programming）同步拉取數(shù)據(jù)的方式，如迭代器模式（Iterator）。而是采用數(shù)據(jù)發(fā)布者同步或異步地推送到數(shù)據(jù)流（Data Streams）的方案。當(dāng)該數(shù)據(jù)流（Data Steams）訂閱者監(jiān)聽到傳播變化時(shí)，立即作出響應(yīng)動(dòng)作。在實(shí)現(xiàn)層面上，Reactive Programming 可結(jié)合函數(shù)式編程簡(jiǎn)化面向?qū)ο笳Z(yǔ)言語(yǔ)法的臃腫性，屏蔽并發(fā)實(shí)現(xiàn)的復(fù)雜細(xì)節(jié)，提供數(shù)據(jù)流的有序操作，從而達(dá)到提升代碼的可讀性，以及減少 Bugs 出現(xiàn)的目的。同時(shí)，Reactive Programming 結(jié)合背壓（Backpressure）的技術(shù)解決發(fā)布端生成數(shù)據(jù)的速率高于訂閱端消費(fèi)的問題。

到此這篇關(guān)于Spring5新特性之Reactive響應(yīng)式編程的文章就介紹到這了,更多相關(guān)Reactive響應(yīng)式編程內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: