Reactive 編程模型有哪些價(jià)值?它的原理是什么?如何正確使用?本文作者將根據(jù)他學(xué)習(xí)和使用的經(jīng)歷，分享 Reactive 的概念、規(guī)范、價(jià)值和原理。歡迎同學(xué)們共同探討、斧正。

Reactive 和 Reactive programming

Reactive 直接翻譯的意思式反應(yīng)式，反應(yīng)性。咋一看，似乎不太好懂。

舉個(gè)例子：在 Excel 里，C 單元格上設(shè)置函數(shù) Sum(A+B)，當(dāng)你改變單元格 A 或者單元格 B 的數(shù)值時(shí)，單元格 C 的值同時(shí)也會(huì)發(fā)生變化。這種行為就是 Reactive。

在計(jì)算機(jī)編程領(lǐng)域，Reactive 一般指的是 Reactive programming。指的是一種面向數(shù)據(jù)流并傳播事件的異步編程范式(asynchronous programming paradigm)。

先舉個(gè)例子大家感受一下：

public static void main(String[] args) { 
   FluxProcessor<Integer, Integer> publisher = UnicastProcessor.create(); 
   publisher.doOnNext(event -> System.out.println("receive event: " + event)).subscribe(); 
  
   publisher.onNext(1); // print 'receive event: 1' 
   publisher.onNext(2); // print 'receive event: 2' 
 }

代碼 1

以上例代碼(使用 Reactor 類庫(kù))為例，publisher 產(chǎn)生了數(shù)據(jù)流 (1,2)，并且傳播給了 OnNext 事件, 上例中 lambda 響應(yīng)了該事件，輸出了相應(yīng)的信息。上例代碼中生成數(shù)據(jù)流和注冊(cè)/執(zhí)行 lambda 是在同一線程中，但也可以在不同線程中。

注：如果上述代碼執(zhí)行邏輯有些疑惑，可以暫時(shí)將 lambda 理解成 callback 就可以了。

Reactive Manifesto

對(duì)于 Reactive 現(xiàn)在你應(yīng)該大致有一點(diǎn)感覺(jué)了，但是 Reactive 有什么價(jià)值，有哪些設(shè)計(jì)原則，估計(jì)你還是有些模糊。這就是 Reactive Manifesto 要解決的疑問(wèn)了。

使用 Reactive 方式構(gòu)建的系統(tǒng)具有以下特征：

即時(shí)響應(yīng)性 (Responsive)

只要有可能， 系統(tǒng)就會(huì)及時(shí)地做出響應(yīng)。即時(shí)響應(yīng)是可用性和實(shí)用性的基石， 而更加重要的是，即時(shí)響應(yīng)意味著可以快速地檢測(cè)到問(wèn)題并且有效地對(duì)其進(jìn)行處理。即時(shí)響應(yīng)的系統(tǒng)專注于提供快速而一致的響應(yīng)時(shí)間， 確立可靠的反饋上限， 以提供一致的服務(wù)質(zhì)量。這種一致的行為轉(zhuǎn)而將簡(jiǎn)化錯(cuò)誤處理、 建立最終用戶的信任并促使用戶與系統(tǒng)作進(jìn)一步的互動(dòng)。

回彈性 (Resilient)

系統(tǒng)在出現(xiàn)失敗時(shí)依然保持即時(shí)響應(yīng)性。這不僅適用于高可用的、 任務(wù)關(guān)鍵型系統(tǒng)——任何不具備回彈性的系統(tǒng)都將會(huì)在發(fā)生失敗之后丟失即時(shí)響應(yīng)性?；貜椥允峭ㄟ^(guò)復(fù)制、 遏制、 隔離以及委托來(lái)實(shí)現(xiàn)的。失敗的擴(kuò)散被遏制在了每個(gè)組件內(nèi)部， 與其他組件相互隔離， 從而確保系統(tǒng)某部分的失敗不會(huì)危及整個(gè)系統(tǒng)，并能獨(dú)立恢復(fù)。每個(gè)組件的恢復(fù)都被委托給了另一個(gè)(外部的)組件， 此外，在必要時(shí)可以通過(guò)復(fù)制來(lái)保證高可用性。(因此)組件的客戶端不再承擔(dān)組件失敗的處理。

彈性 (Elastic)

系統(tǒng)在不斷變化的工作負(fù)載之下依然保持即時(shí)響應(yīng)性。反應(yīng)式系統(tǒng)可以對(duì)輸入(負(fù)載)的速率變化做出反應(yīng)，比如通過(guò)增加或者減少被分配用于服務(wù)這些輸入(負(fù)載)的資源。這意味著設(shè)計(jì)上并沒(méi)有爭(zhēng)用點(diǎn)和中央瓶頸， 得以進(jìn)行組件的分片或者復(fù)制， 并在它們之間分布輸入(負(fù)載)。通過(guò)提供相關(guān)的實(shí)時(shí)性能指標(biāo)， 反應(yīng)式系統(tǒng)能支持預(yù)測(cè)式以及反應(yīng)式的伸縮算法。這些系統(tǒng)可以在常規(guī)的硬件以及軟件平臺(tái)上實(shí)現(xiàn)成本高效的彈性。

消息驅(qū)動(dòng) (Message Driven)

反應(yīng)式系統(tǒng)依賴異步的消息傳遞，從而確保了松耦合、隔離、位置透明的組件之間有著明確邊界。這一邊界還提供了將失敗作為消息委托出去的手段。使用顯式的消息傳遞，可以通過(guò)在系統(tǒng)中塑造并監(jiān)視消息流隊(duì)列， 并在必要時(shí)應(yīng)用回壓， 從而實(shí)現(xiàn)負(fù)載管理、 彈性以及流量控制。使用位置透明的消息傳遞作為通信的手段， 使得跨集群或者在單個(gè)主機(jī)中使用相同的結(jié)構(gòu)成分和語(yǔ)義來(lái)管理失敗成為了可能。非阻塞的通信使得接收者可以只在活動(dòng)時(shí)才消耗資源， 從而減少系統(tǒng)開(kāi)銷。

??

注：

• 上面描述有很多專有名詞，可能有些疑惑，可以看下相關(guān)名詞解釋。
• 為什么使用 Reactive 方式構(gòu)建的系統(tǒng)會(huì)具有以上價(jià)值, 我稍后在 Reactor 章節(jié)中介紹。

Reactive Stream

知道了 Reactive 的概念，特征和價(jià)值后，是否有相關(guān)的產(chǎn)品或者框架來(lái)幫助我們構(gòu)建 Reactive 式系統(tǒng)呢?在早些時(shí)候有一些類庫(kù) (Rxjava 1.x, Rx.Net) 可以使用，但是規(guī)范并不統(tǒng)一，所以后來(lái) Netfilx, Pivotal 等公司就制定了一套規(guī)范指導(dǎo)大家便于實(shí)現(xiàn)它(該規(guī)范也是受到早期產(chǎn)品的啟發(fā))，這就是 Reactive Stream 的作用。

Reactive Stream 是一個(gè)使用非阻塞 back pressure(回壓)實(shí)現(xiàn)異步流式數(shù)據(jù)處理的標(biāo)準(zhǔn)。目前已經(jīng)在 JVM 和 JavaScript 語(yǔ)言中實(shí)現(xiàn)同一套語(yǔ)意的規(guī)范;以及嘗試在各種涉及到序列化和反序列化的傳輸協(xié)議(TCP, UDP, HTTP and WebSockets)基礎(chǔ)上，定義傳輸 reactive 數(shù)據(jù)流的網(wǎng)絡(luò)協(xié)議。

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

Reactive Streams 解決的問(wèn)題場(chǎng)景

當(dāng)遇到未預(yù)料數(shù)據(jù)流時(shí)，依然可以在可控資源消耗下保持系統(tǒng)的可用性。

Reactive Streams 的目標(biāo)

控制在一個(gè)異步邊界的流式數(shù)據(jù)交換。例如傳遞一個(gè)數(shù)據(jù)到另外一個(gè)線程或者線程池，確保接收方?jīng)]有 buffer(緩存)任意數(shù)量的數(shù)據(jù)。而 back pressure(回壓)是解決這種場(chǎng)景的不可或缺的特性。

Reactive Streams 規(guī)范適用范圍

此標(biāo)準(zhǔn)只描述通過(guò)回壓來(lái)實(shí)現(xiàn)異步流式數(shù)據(jù)交換的必要的行為和實(shí)體，最小接口，例如下方的 Publisher, Subscriber。Reactive Streams 只關(guān)注在這些組件之間的流式數(shù)據(jù)中轉(zhuǎn)，并不關(guān)注流式數(shù)據(jù)本身的組裝，分割，轉(zhuǎn)換等行為, 例如 map, zip 等 operator。Reactive Streams 規(guī)范包括：

Publisher

產(chǎn)生一個(gè)數(shù)據(jù)流(可能包含無(wú)限數(shù)據(jù)), Subscriber 們可以根據(jù)它們的需要消費(fèi)這些數(shù)據(jù)。

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

Subscriber

Publisher 創(chuàng)建的元素的接收者。監(jiān)聽(tīng)指定的事件，例如 OnNext, OnComplete, OnError 等。

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

Subscription

是 Publisher 和 Subscriber 一對(duì)一的協(xié)調(diào)對(duì)象。Subscriber 可以通過(guò)它來(lái)向 Publisher 取消數(shù)據(jù)發(fā)送或者 request 更多數(shù)據(jù)。

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

Processor

同時(shí)具備 Publisher 和 Subscriber 特征。代碼1中 FluxProcessor 既可以發(fā)送數(shù)據(jù)(OnNext)，也可以接收數(shù)據(jù) (doOnNext)。

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

為什么規(guī)范強(qiáng)調(diào)使用非阻塞異步方式而不是阻塞同步方式?

• 同步方式一般通過(guò)多線程來(lái)提高性能，但系統(tǒng)可創(chuàng)建的線程數(shù)是有限的，且線程多以后造成線程切換開(kāi)銷。
• 同步方式很難進(jìn)一步提升資源利用率。
• 同步調(diào)用依賴的系統(tǒng)出現(xiàn)問(wèn)題時(shí)，自身穩(wěn)定性也會(huì)受到影響。

實(shí)現(xiàn)非阻塞的方式有很多種，為什么規(guī)范會(huì)選擇上述的實(shí)現(xiàn)方式呢?

Thread

• thread 不是非常輕量(相比下面幾種實(shí)現(xiàn)方案)。
• thread 數(shù)量是有限的，最終可能會(huì)成為主要瓶頸。
• 有一些平臺(tái)可能不支持多線程。例如：JavaScript。
• 調(diào)試，實(shí)現(xiàn)上有一定復(fù)雜性。

Callback

• 多層嵌套 callback 比較復(fù)雜，容易形成"圣誕樹(shù)" (callback hell)。
• 錯(cuò)誤處理比較復(fù)雜。
• 多用于 event loop 架構(gòu)的語(yǔ)言中，例如：JavaScript。

Future

• 無(wú)法邏輯組合各種行為，支持業(yè)務(wù)場(chǎng)景有限。
• 錯(cuò)誤處理依然復(fù)雜。

Reactive Extensions (Rx)

• 和 Future 很相似。Future 可以認(rèn)為返回一個(gè)獨(dú)立的元素，而 Rx 返回一個(gè)可以被訂閱的 Stream。
• 多平臺(tái)支持同一套規(guī)范。
• 同一套 API 同時(shí)支持異步、同步。
• 錯(cuò)誤處理方便。

Coroutines

• kotlin coroutine 和 goroutine 在語(yǔ)法層面上提供異步支持, 而且比Rx更簡(jiǎn)潔，但無(wú)法跨多個(gè)語(yǔ)言平臺(tái)形成統(tǒng)一的規(guī)范。

Reactive 的實(shí)現(xiàn)原理個(gè)人認(rèn)為還是回調(diào)，kotlin 協(xié)程實(shí)現(xiàn)原理同樣也是回調(diào)。但實(shí)現(xiàn)回掉的方式不一樣。一個(gè)是通過(guò)事件傳播, 一個(gè)是通過(guò)狀態(tài)機(jī)。但 cooutine 編程的易用性明顯強(qiáng)于 Rx，后面有空我會(huì)專門寫(xiě)篇文章介紹 kotlin coroutine 的實(shí)現(xiàn)原理。

Reactor

有了 Reactive Stream 這個(gè)規(guī)范，就會(huì)有相應(yīng)實(shí)現(xiàn)該規(guī)范的類庫(kù)。Reactor 就是其中之一。

Reactor 是遵守 Reactive Stream 規(guī)范構(gòu)建非阻塞應(yīng)用的 Java 語(yǔ)言 Reactive 類庫(kù)，已經(jīng)在 spring 5 中集成，與他相似的類庫(kù)有 RxJava2, RxJs, JDK9 Flow 等。

阿里內(nèi)部的 Faas 系統(tǒng)目前使用 Reactor 來(lái)構(gòu)建整個(gè)系統(tǒng)，包括函數(shù)應(yīng)用和各種核心應(yīng)用(邏輯架構(gòu))。根據(jù)我們壓測(cè)結(jié)果顯示，使用 Reactive 方式構(gòu)建的系統(tǒng)確實(shí)會(huì)有這些特點(diǎn)：

• 回彈性 (Resilient)：當(dāng)函數(shù)出現(xiàn)嚴(yán)重超時(shí)時(shí) (rt >= 10s)，函數(shù)上游的 broker, gateway 應(yīng)用幾乎無(wú)任何影響。
• 及時(shí)響應(yīng)性：不管是高并發(fā)場(chǎng)景(資源足夠)，還是正常場(chǎng)景，RT 表現(xiàn)一致。

另外從原理上，我認(rèn)為資源利用率和吞吐量也會(huì)高于非反應(yīng)式的應(yīng)用。

為什么 Reactive 的架構(gòu)系統(tǒng)有這些特點(diǎn)?

阿里內(nèi)部的 Faas 系統(tǒng)主要做了兩件事情：

1. 涉及到 IO 的地方幾乎全異步化。例如中間件(HSF, MetaQ 等提供異步 API)調(diào)用。
2. IO 線程模型變化。使用較少(一般 CPU 核數(shù))線程處理所有的請(qǐng)求。

傳統(tǒng) Java 應(yīng)用 IO 線程模型

參考 Netty 中 Reactor IO (worker thread pool) 模型，下方偽代碼(kotlin)進(jìn)行了簡(jiǎn)化。

??

// 非阻塞讀取客戶端請(qǐng)求數(shù)據(jù)(in), 讀取成功后執(zhí)行l(wèi)ambda. 
 inChannel.read(in) { 
     workerThreadPool.execute{ 
       // 阻塞處理業(yè)務(wù)邏輯(process), 業(yè)務(wù)邏輯在worker線程池中執(zhí)行，同步執(zhí)行完后，再向客戶端返回輸出(out) 
       val out = process(in) 
       outChannel.write(out) 
     }   
 }

Reactive 應(yīng)用 IO 線程模型

IO 線程也可以執(zhí)行業(yè)務(wù)邏輯 (process)，可以不需要 worker 線程池。

??

// 非阻塞讀取客戶端請(qǐng)求數(shù)據(jù)(in), 讀取成功后執(zhí)行l(wèi)ambda 
 inChannel.read(in) { 
     // IO線程執(zhí)行業(yè)務(wù)邏輯(process),  然后向客戶端返回輸出(out). 這要求業(yè)務(wù)處理流程必須是非阻塞的. 
     process(in){ out-> 
         outChannel.write(out) { 
             // this lambda is executed when the writing completes 
         ... 
         } 
     } 
 }

如何開(kāi)始 Reactive Programing

以 Reactive 方式構(gòu)建的系統(tǒng)有很多值得學(xué)習(xí)和發(fā)揮價(jià)值的地方，但坦白講 Reactive programing 方式目前接受程度并不高。特別是使用 Java 語(yǔ)言開(kāi)發(fā)同學(xué)，我個(gè)人也感同身受，因?yàn)檫@和 Java 面向命令控制流程的編程思維方式有較大差異。所以這里以 Reactor (Java) 學(xué)習(xí)為例：

• Reactor 基礎(chǔ)文檔
• Reactive Streams 規(guī)范文檔
• Operator

總結(jié)

反應(yīng)式的系統(tǒng)有很多優(yōu)點(diǎn)，但是完整構(gòu)建反應(yīng)式的系統(tǒng)卻并不容易。不僅僅是語(yǔ)言上的差異，還有一些組件就不支持非阻塞式的調(diào)用方式，例如：JDBC。但是有一些開(kāi)源組織正在推動(dòng)這些技術(shù)進(jìn)行革新，例如：R2DBC。另外，為了方便構(gòu)建反應(yīng)式系統(tǒng)，一些組織/個(gè)人適配了一些主流技術(shù)組件 reactor-core, reactor-netty, reactor-rabbimq, reactor-kafka 等，來(lái)方便完整構(gòu)建反應(yīng)式系統(tǒng)。

當(dāng)你的系統(tǒng)從底層到上層，從系統(tǒng)內(nèi)部到依賴外部都變成了反應(yīng)式，這就形成了 Reactive 架構(gòu)。

這種架構(gòu)價(jià)值有多大?未來(lái)可期。

參考

??https://www.reactivemanifesto.org/??

??https://www.reactive-streams.org/??

??https://kotlinlang.org/docs/tutorials/coroutines/async-programming.html??

??https://projectreactor.io/docs/core/release/reference/index.html??