1 什么是響應式編程

一句話總結(jié)：響應式編程是一種編程范式，通用和專注于數(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.

翻譯：

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

舉例：

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

響應式編程最初是為了簡化交互式用戶界面的創(chuàng)建和實時系統(tǒ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.

翻譯：

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

2.2 為什么是Reactor

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

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

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

換句話說，

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

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

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

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

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

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

結(jié)構(gòu)對應：

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

模式對應：

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

流程：

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

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:\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 測試

@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響應式編程？

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

解釋：

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

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

最新評論