Reactive Programming入門概念詳解

更新時間：2022年09月28日 16:24:58 作者：暗里著迷for_you

這篇文章主要為大家介紹了Reactive Programming入門概念詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步，早日升職加薪

正文

為了應對高并發(fā)環(huán)境下的服務端編程，xx提出了一個實現(xiàn)異步編程的方案 -Reactive Programming，中文名稱反應式編程。反應式編程(reactive programming)并不是一個新的概念，也不是一個新的技術，很早之前就被提出來了。

先從幾個概念入門說起：

Reactive Programming
Reactive Streams
Reactor
Webflux

Reactive Programming

反應式編程 (reactive programming) 是一種基于數(shù)據(jù)流 (data stream) 和變化傳遞 (propagation of change) 的聲明式 (declarative) 的編程范式。

響應式編程使用三個核心概念：數(shù)據(jù)流，變化傳遞和異步觀察。

換句話說：使用異步數(shù)據(jù)流進行編程，這意味著可以在編程語言中很方便地表達靜態(tài)或動態(tài)的數(shù)據(jù)流，而相關的計算模型會自動將變化的值通過數(shù)據(jù)流進行傳播。反應式編程提高了代碼的抽象級別，可以只關注定義了業(yè)務邏輯的那些相互依賴的事件。

Reactive Streams

Reactive Streams ：The purpose of Reactive Streams is to provide a standard for asynchronous stream processing with non-blocking backpressure。

Reactive Streams 的目的是為具有非阻塞背壓的異步流處理提供標準。

對于Java程序員,Reactive Streams是一個API。Reactive Streams為我們提供了Java中的Reactive Programming的通用API。

Reactive Streams API的范圍是找到一組最小的接口，方法和協(xié)議，這些接口，方法和協(xié)議將描述必要的操作和實體，從而實現(xiàn)具有非阻塞背壓的異步數(shù)據(jù)流。

Reactive Streams API中如下四個接口：（后面文章詳細）

Publisher 發(fā)布者（生產(chǎn)者）

public interface Publisher<T> {    
    public void subscribe(Subscriber<? super T> s);
}

Subscriber訂閱者（消費者）

public interface Subscriber<T> {    
    public void onSubscribe(Subscription s);    
    public void onNext(T t);    
    public void onError(Throwable t);    
    public void onComplete();
}

Subscription 訂閱

public interface Subscription {   
    public void request(long n);    
    public void cancel();
}

Processor 處理者

public interface Processor<T, R> extends Subscriber<T>, Publisher<R> {
}

Spring Reactor

Reactor 是第四代響應式庫，基于Reactive Streams規(guī)范，用于在 JVM 上構建非阻塞應用程序。是Spring提供的非阻塞式響應式編程框架，實現(xiàn)了Reactive Streams規(guī)范而已

reactor-core: Reactor 完全無阻塞，并提供高效的管理，它直接與Java的函數(shù)式API,CompletableFuture,Stream配合使用
- [0|1| N]序列：Reactore提供了兩個響應式和可組合的API,Flux 和 Mono。

Flux 表示的是包含 0 到 N 個元素的異步序列。在該序列中可以包含三種不同類型的消息通知：正常的包含元素的消息、序列結束的消息和序列出錯的消息。當消息通知產(chǎn)生時，訂閱者中對應的方法 onNext(), onComplete()和 onError()會被調(diào)用。Mono 表示的是包含 0 或者 1 個元素的異步序列。該序列中同樣可以包含與 Flux 相同的三種類型的消息通知。Flux 和 Mono 之間可以進行轉換。對一個 Flux 序列進行計數(shù)操作，得到的結果是一個 Mono對象。把兩個 Mono 序列合并在一起，得到的是一個 Flux 對象。

Non-Blocking IO: Reactor 為HTTP(包括websocket),TCP和UDP提供背壓就緒的網(wǎng)絡引擎。

webflux

Spring 產(chǎn)品組合提供了兩個并行堆棧。一種是基于帶有 Spring MVC 和 Spring Data 結構的 Servlet API。另一個是完全反應式堆棧，它利用了 Spring WebFlux 和 Spring Data 的反應式存儲庫。如下技術棧：Spring官方提供

數(shù)據(jù)庫：以一種反應的方式訪問和處理數(shù)據(jù)是非常重要的。MongoDB、Redis和Cassandra都有原生的Spring數(shù)據(jù)響應支持。許多關系數(shù)據(jù)庫(Postgres、Microsoft SQL Server、MySQL、H2和GoogleSpanner)通過R2DBC提供了反應性支持。

在消息傳遞的世界中，Spring Cloud Stream還支持對RabbitMQ和Kafka等平臺的反應性訪問。