反應式架構(1)：基本概念介紹 頂

• 2019 年 12 月 12 日
• 筆記

       淘寶從2018年開始對整體架構進行反應式升級， 取得了非常好的成績。其中『猜你喜歡』應用上限 QPS 提升了 96%，同時機器數量縮減了一半；另一核心應用『我的淘寶』實際線上響應時間下降了 40% 以上。PayPal憑藉其基於Akka構建的反應式平台squbs，僅使用8台2vCPU虛擬機，每天可以處理超過10億筆交易，與基於Spring實現的老系統相比，代碼量降低了80%，而性能卻提升了10倍。能夠取得如此好的成績，人們不禁要問反應式到底是什麼？ 其實反應式並不是一個新鮮的概念，它的靈感來源最早可以追溯到90年代，但是直到2013年，Roland Kuhn等人發佈了《反應式宣言》後才慢慢被人熟知，繼而在2014年迎來爆髮式增長，比較有意思的是，同時迎來爆髮式增長的還有領域驅動設計(DDD)，原因是2014年3月25日，Martin Fowler和James Lewis向大眾介紹了微服務架構，而反應式和領域驅動是微服務架構得以落地的有力保障。緊接着各種反應式編程框架相繼進入大家視野，如RxJava、Akka、Spring Reactor/WebFlux、Play Framework和未來的Dubbo3等，阿里內部在做反應式改造時也孵化了一些反應式項目，包括AliRxObjC、RxAOP和AliRxUtil等。 從目前的趨勢看來，反應式概念將會逐漸深入人心， 並且將引領下一代技術變革。

       本文將向大家介紹什麼是反應式，以及為什麼要採用反應式架構，並且通過一個編程示例，深入分析傳統的編程方式會帶來哪些問題和挑戰，以及如何做異步化改造，順利邁出反應式架構演進的第一步。

1 什麼是反應式？

1.1 反應式介紹

       為了直觀地了解什麼是反應式，我們先從一個大家都比較熟悉的類比開始。首先打開Excel，在B、C、D三列輸入如下公式：

       B、C和D三列每個單元格的值均依賴其左側的單元格，當我們在A列依次輸入1、2和3時，變化會自動傳遞到了B、C和D三列，並觸發相應狀態變更，如下圖：

       我們可以把A列從上到下想像成一個數據流，每一個數據到達時都會觸發一個事件，該事件會被傳播到右側單元格，後者則會處理事件並改變自身的狀態。這一系列流程其實就是反應式的核心思想。

       通過這個例子，你應該能感受到反應式的核心是數據流(data stream)， 下面我們再來看一個例子。我們很多人每天都會坐地鐵上下班，地鐵每兩分鐘一班，並且同一條軌道會被很多地鐵共享，你會不會因為擔心追尾，而不敢坐首尾兩節車廂呢？ 其實如果採用反應式架構構建地鐵系統，就無需擔心追尾問題。在反應式系統中，每輛地鐵都會實時將自己的速度和位置等狀態信息通知給上下游的其他地鐵，同時也會實時的接收其他地鐵的狀態信息，並實時做出反饋。例如當發現下游地鐵突然意外減速，則立即調整自身速度，並將減速事件通知到上游地鐵，如此，整條軌道上的所有地鐵形成一種回壓機制(back pressure)，根據上下游狀態自動調整自身速度。 下面我們來看下維基百科關於反應式編程的定義：

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

       從上面的定義中，我們可以看出反應式編程的核心是數據流以及變化傳遞。維基百科給出的定義比較通用，具有普適性，沒有區分數據流的同步和異步模式， 更準確地說，異步數據流(asynchronous data stream)或者說反應式流(reactive stream)才是反應式編程的最佳實踐。細心的讀者會發現，講了這麼多，這不就是觀察者模式(Observer Pattern)嘛！ 其實這個說法並不準確，其實反應式並不是指具體的技術，而是指一些架構設計原則， 觀察者模式是實現反應式的一種手段，在接下來的反應式流(Reactive Stream)一節，我們會發現反應式流基於觀察者模式擴展了更多的功能，更強大也更易用。

1.2 反應式歷史

       早在1985年，David Harel 和 Amir Pnueli 就發表了《反應式系統的開發》論文，在論文中，他們採用二分法對複雜的計算過程進行歸納，提出了轉換式(transformative)與反應式(reactive)系統。其中反應式系統就是指能夠持續地與環境進行交互，並且及時地進行響應。例如視頻監控系統會持續監測， 併當有陌生人闖入時立刻觸發警報。

表1 反應式歷史

圖1 谷歌搜索趨勢

       從Google搜索趨勢上可以看出，從2013年6月份開始，反應式編程的搜素趨勢出現了爆髮式增長，原因是2013年6月反應式宣言發佈了第一個版本。

1.3 ReactiveX 介紹

       ReactiveX是Reactive Extensions的縮寫，一般簡寫為Rx，最初是LINQ的一個擴展，由微軟的架構師Erik Meijer領導的團隊開發，在2012年11月開源。Rx是一個編程模型，目標是提供一致的編程接口，幫助開發者更方便的處理異步數據流。Rx支持幾乎全部的流行編程語言，大部分語言庫由ReactiveX這個組織負責維護，比較流行的有RxJava/RxJS/Rx.NET/Rx.Scala/ Rx.Swift，社區網站是http://reactivex.io/。

1.4 反應式宣言

       2013年6月，Roland Kuhn等人發佈了《反應式宣言》， 該宣言定義了反應式系統應該具備的一些架構設計原則。符合反應式設計原則的系統稱為反應式系統。根據反應式宣言， 反應式系統需要具備即時響應性(Responsive)、回彈性(Resilient)、彈性(Elastic)和消息驅動(Message Driven)四個特質，以下內容摘自反應式宣言官網， 描述比較抽象，大家不必糾結細節，了解即可。

• 即時響應性(Responsive)。系統應該對用戶的請求即時做出響應。即時響應是可用性和實用性的基石， 而更加重要的是，即時響應意味着可以快速地檢測到問題並且有效地對其進行處理。
• 回彈性(Resilient)。 系統在出現失敗時依然能保持即時響應性， 每個組件的恢復都被委託給了另一個外部的組件， 此外，在必要時可以通過複製來保證高可用性。 因此組件的客戶端不再承擔組件失敗的處理。
• 彈性(Elastic)。 系統在不斷變化的工作負載之下依然保持即時響應性。 反應式系統可以對輸入負載的速率變化做出反應，比如通過橫向地伸縮底層計算資源。 這意味着設計上不能有中央瓶頸， 使得各個組件可以進行分片或者複製， 並在它們之間進行負載均衡。
• 消息驅動(Message Driven)。反應式系統依賴異步的消息傳遞，從而確保了松耦合、隔離、位置透明的組件之間有着明確邊界。 這一邊界還提供了將失敗作為消息委託出去的手段。 使用顯式的消息傳遞，可以通過在系統中塑造並監視消息流隊列， 並在必要時應用回壓， 從而實現負載管理、 彈性以及流量控制。 使用位置透明的消息傳遞作為通信的手段， 使得跨集群或者在單個主機中使用相同的結構成分和語義來管理失敗成為了可能。 非阻塞的通信使得接收者可以只在活動時才消耗資源， 從而減少系統開銷。

1.5 Reactive Streams

       反應式宣言僅闡述了設計原則，並沒有給出具體的實現規範，導致每個反應式框架都各自實現了一套自己的API規範，且相互之間無法互通。為了解決這個問題，Reactive Streams規範應運而生。

       Reactive Streams的目標是定義一組最小化的異步流處理接口，使得在不同框架之間，甚至不同語言之間實現交互性。Reactive Streams規範包含了4個接口，7個方法，43條規則以及一套用於兼容性測試的標準套件TCK(The Technology Compatibility Kit)。該規範已經成為了業界標準， 並且在Java 9中已經實現，對應的實現接口為java.util.concurrent.Flow。 有一點需要提醒的是，雖然Java 9已經實現了Reactive Streams，但這並不意味着像RxJava、Reactor、Akka Streams這些流處理框架就沒有意義了，事實上恰恰相反。Reactive Streams的目的在於增強不同框架之間的交互性，提供的是一組最小功能集合，無法滿足我們日常的流處理需求，例如組合、過濾、緩存、限流等功能都需要額外實現。流處理框架的目的就在於提供這些額外的功能實現，並通過Reactive Streams規範實現跨框架的交互性。

       舉個例子來說，MongoDB的Java驅動實現了Reactive Streams規範， 開發者使用任何一個流處理框架，僅需要幾行代碼即可實時監聽數據庫的變化。例如下面是基於Akka Stream的實現代碼：

mongo    .collection("users")    .watch()    .toSource    .groupedWithin(10, 1.second)    .throttle(1, 1.second)    .runForeach { docs =>       // 處理增量數據  }

     上面的幾行代碼實現了如下功能：

• 將接收到的流數據進行緩衝以方便批處理，滿足以下任一條件便結束緩衝並向後傳遞
  • 緩衝滿10個元素
  • 緩衝時間超過了1000毫秒
• 對緩衝後的元素進行流控，每秒只允許通過1個元素

1.6 小結

       本章首先通過形象的例子讓大家對反應式系統有一個直觀的認知，然後帶領大家一起回顧了反應式的發展歷史，最後向大家介紹了三個反應式項目，包括ReactiveX、反應式宣言和Reactive Streams。 ReactiveX是反應式擴展，旨在為各個編程語言提供反應式編程工具。反應式宣言站在一個更高的角度，使用抽象語言向大家描述什麼是反應式系統，以及實現反應式系統應該遵循的一些設計原則。Reactive Streams規範的目的在於提高各個反應式框架之間的交互性，本身並不適合作為開發框架直接使用，開發者應該選擇一個成熟的反應式框架，並通過Reactive Streams規範與其它框架實現交互。

2 為什麼需要反應式？

2.1 命令式編程 VS 聲明式編程

       實際上我們絕大多數程序員都在使用傳統的命令式編程，這也是計算機的工作方式。命令式編程就是對硬件操作的抽象， 程序員需要通過指令，精確的告訴計算機幹什麼事情。這也是編程工作中最枯燥的地方，程序員需要耗盡腦汁，將複雜、易變的業務需求翻譯成精確的計算機指令。

       聲明式編程是解決程序員的利器，聲明式編程更關注我想要什麼(What)而不是怎麼去做(How)。SQL是最典型的聲明式語言，我們通過SQL描述想要什麼，最終由數據庫引擎執行SQL語句並將結果返回給我們。

SELECT COUNT(*)  FROM USER u  WHERE u.age > 30

       1.5節使用Akka Stream實現監聽MongoDB的代碼也是典型的聲明式編程，如果採用命令式方式重寫， 不僅費時費力，而且還會導致代碼量暴增，最重要的是要通過更多的單元測試保證實現的正確性。

       反應式架構推薦使用聲明式編程， 使用更接近自然語言的方式描述業務邏輯， 代碼清晰易懂並且富有表達力， 最重要的是大大降低了後期維護成本。

2.2 同步編程 VS 異步編程

       當談到同步與異步時，就不得不提一下阻塞與非阻塞的概念，因為這兩組概念很容易混淆。導致混淆的原因是它們在描述同一個東西，但是關注點不同。 阻塞與非阻塞關注方法執行時當前線程的狀態，而同步與異步則關注方法調用結果的通知機制。因為是從不同角度描述方法的調用過程，所以這兩組概念也可以相互組合，即將線程狀態和通知機制進行組合。例如JDK1.3及以前的BIO是同步阻塞模式，JDK1.4發佈的NIO是同步非阻塞模式，JDK1.7發佈的NIO.2是異步非阻塞模式。

       跟命令式編程一樣，同步編程也是目前被廣泛採用的傳統編程方式。同步編程的優點是代碼簡單並且容易理解，代碼按照先後順序依次執行；缺點是CPU利用率非常低，大部分時間都白白浪費在了IO等待上。

       異步編程通過充分利用CPU資源並行執行任務， 在執行時間和資源利用率上遠遠高於同步方式。舉個例子來說，對於一個10核服務器，使用同步方式抓取10個網頁，每個網頁耗時1秒，則總耗時為10秒；如果採用異步方式，10個抓取任務分別在各自的線程上執行，總耗時只有1秒。 構建反應式系統並非易事，尤其是針對遺留系統進行改造，這將會是一個較為漫長的過程。反應式架構的核心思想是異步非阻塞的反應式流，作為過渡階段，我們可以選擇先對系統進行完全異步化重構，為進一步向反應式架構演進奠定基礎。接下來，我們將先分析一個傳統的同步示例，然後針對該示例進行異步化重構。

2.3 同步編程示例

       假設我們要實現一個查詢手機套餐餘額的方法， 該方法接受一個手機號參數，返回該手機號的套餐餘額信息， 包括剩餘通話時間、剩餘短訊數量和剩餘網絡流量。 由於查詢套餐餘額需要連續發起三次同步阻塞的數據庫查詢請求，所以在實現中需要利用緩存提高讀取性能， 代碼如下：

private PhonePlanCache cache;    public PhonePlan retrievePhonePlan(String phoneNo) {      PhonePlan plan = cache.get(phoneNo);      if (plan != null) {          return plan;      }        Long leftTalk = readLeftTalk(phoneNo);      Long leftText = readLeftText(phoneNo);      Long leftData = readLeftData(phoneNo);        return new PhonePlan(leftTalk, leftText, leftData);  } 

       首先我們檢查是否可以直接從緩存中讀取套餐餘額信息，如果可以則直接返回， 否則連續發起三次同步阻塞的遠程調用， 從數據庫中依次讀取通話餘額、短訊餘額和流量餘額。代碼邏輯非常簡單，但是由於同步阻塞代碼對線程池依賴非常嚴重，接下來我們還需要根據SLA估算線程池和連接池大小。估算的過程並不容易，好在我們有利特爾法則。

       1954年， John Little基於等候理論提出了利特爾法則(Little's law)： 在一個穩定的系統中，系統可以同時處理的請求數量L， 等於請求到達的平均速度 λ 乘以請求的平均處理時間W， 即：

L = λ * W

       這個法則同樣可以用來計算線程池和連接池大小。 例如系統每秒接收1000個請求，每個請求的平均處理時間是10ms， 則合適的數據庫連接池大小應該為10。 也就是說系統可以同時處理10個請求。 從長時間來看，系統平均會有10個線程在等待數據庫連接上的響應。 但是需要注意的是，利特爾法則只適用於一個穩定系統， 無法處理峰值情況， 而通常系統請求數量的峰值會比平均值高很多。假設為了應付峰值情況，我們將線程池大小調整為50， 由於連接池大小仍為10，所以會導致大量線程在等待可用連接， 我們需要再次增大連接池大小以改善系統性能。通常經過如此反覆調整後的參數已經嚴重偏離了利特爾法則， 導致系統性能嚴重下降，在高並發場景下，如果網絡稍有抖動或數據庫稍有延遲，則會導致瞬間積壓大量請求， 如果沒有有效的應對措施，系統將面臨癱瘓風險。

2.4 同步編程面臨的挑戰

       傳統應用通常基於Servlet容器進行部署，而Servlet是基於線程的請求處理模型。從上文的討論中我們發現，通常需要設置一個較大的線程池以獲得較好的性能，較大的線程池會導致以下三個問題：

• 額外的內存開銷。 在Java中，每個線程都有自己的棧空間，默認是1MB。如果設置線程池大小為200，則應用在啟動時至少需要200M內存，一方面造成了內存浪費，另一方面也導致應用啟動變慢。試想一下，如果同時部署1000個節點，這些問題將會被放大1000倍。
• CPU利用率低。 有兩個方面原因會導致極低的CPU利用率。一方面是在Oracle JDK 1.2版本之後，所有平台的JVM實現都使用1:1線程模型(Solaris是個特例)，這意味着一個Java線程會被映射到一個輕量級進程上，而有效的輕量級進程數量取決於CPU的個數以及核數。如果Java的線程數量遠大於有效的輕量級進程數量，則頻繁的線程上限文切換會浪費大量CPU時間； 另一方面，由於傳統的遠程操作或IO操作均為阻塞操作，會導致執行線程被掛起從而無法執行其他任務，大大降低了CPU的利用率。
• 資源競爭激烈。 當增大線程池後，其他的共享資源便會成為性能瓶頸，如數據庫連接池資源。如果存在共享資源瓶頸，即使設置再大的線程池，也無法有效地提升性能。此時會導致多個線程競爭數據庫連接， 使得數據庫連接成為系統瓶頸。

    除了上面這些問題，同步編程還會深刻地影響到我們的架構。

    假設我們準備開發一個單點登錄微服務，微服務框架使用 Dubbo 2.x，該版本尚未支持反應式編程，微服務接口之間調用仍然是同步阻塞方式。 假設我們需要實現如下兩個接口：

• 用戶登錄接口
• 令牌驗證接口

    對於用戶登錄接口，由於需要多次訪問數據庫或緩存，並且需要使用Argon2等慢哈希算法進行密碼校驗，導致平均響應時間較長，約為500毫秒。而對於令牌驗證接口，由於只需要做簡單的簽名校驗，所以平均響應時間較短，約為5毫秒。 假設由於業務需要，用戶登錄接口的性能指標只需要達到1000tps即可，而令牌驗證接口的性能指標則需要達到100,000tps。

     通常來說，這兩個接口會在同一個微服務類中實現，也通常會被發佈到同一個容器中對外提供服務。為了滿足業務需要，我們先來算一下需要多少硬件成本？ 為了簡化討論，我們認為令牌驗證接口無需硬件成本，只關注用戶登錄接口即可。根據利特爾法則， 總線程數量(L) = TPS(λ)*平均響應時間(W)， 即：

總線程數量(L) = (1000*0.5) = 500

     假設每個計算節點配置為4C8G， 那麼一共需要 (500/4)=125台計算節點。 區區的1000tps竟然需要125台計算節點！你以為這就完了嗎？ 1000tps只是日常的請求壓力，如果考慮峰值情況呢？假設峰值請求是10, 000tps，並且會持續10秒， 那麼在這10秒內系統也可以看做是穩定狀態， 那麼根據利特爾法則，就需要部署1250台計算節點。 還有更壞的情況，如果某個節點由於數據庫延遲或網絡抖動等情況，導致用戶登錄請求積壓，則用戶登錄請求會耗盡所有請求處理線程，導致原本可以快速響應的令牌驗證請求無法被及時處理，而令牌驗證接口的tps是100,000，這意味着1秒鐘就會積壓100,000個令牌驗證請求， 系統已經處在危險邊緣，隨時都會崩潰。

     為了解決令牌驗證接口的快速響應問題，我們只能調整架構，將登陸和驗證拆分成兩個單獨的微服務，並且各自部署到獨立的容器中。這樣是不是就萬事大吉了呢？很不幸，單點登錄迎來了一個新需求，針對員工賬戶需要遠程調用LDAP進行認證， 而遠程調用LDAP也是一個同步阻塞操作，這意味着每一個LDAP遠程調用都會掛起一個線程，大量的遠程調用也會耗盡所有線程，這些被掛起的線程啥都不做，就在那傻傻的等待遠程響應。這其實就是微服務調用鏈雪崩的罪魁禍首。兩個微服務之間調用已經如此棘手了，那如果調用鏈上有10個甚至更多的微服務調用呢？ 那將是一場噩夢！

     其實所有問題的根源都可以歸結為傳統的同步阻塞編程方式。尤其是在微服務場景下，隨着調用鏈長度的不斷增長，風險也將越來越高， 其中任何一個節點同步阻塞操作都會導致其下游所有節點線程被阻塞，如果問題節點的請求產生積壓，則會導致所有下游節點線程被耗盡，這就是可怕的雪崩。

2.5 異步編程示例

     我們說異步編程通常是指異步非阻塞的編程方式，即要求系統中不能有任何阻塞線程的代碼。在現實情況下，想實現完全的異步非阻塞非常困難， 因為還有很多第三方的庫或驅動仍然採用同步阻塞的編程方式。我們需要為這些庫或驅動指定獨立的線程池，以免影響到其他服務接口。

     利用Java 8提供的CompletableFuture和Lambda兩個特性，我們對2.2節的示例進行異步化改造，改造後代碼如下：

private PhonePlanCache cache;    public CompletableFuture<PhonePlan> retrievePhonePlan(String phoneNo) {      PhonePlan cachedPlan = cache.get(phoneNo);      if (cachedPlan != null) {          return CompletableFuture.completedFuture(cachedPlan);      }        CompletableFuture<Long> leftTalkFuture = readLeftTalk(phoneNo);      CompletableFuture<Long> leftTextFuture = readLeftText(phoneNo);      CompletableFuture<Long> leftDataFuture = readLeftData(phoneNo);        CompletableFuture<PhonePlan> planFuture =          leftTalkFuture.thenCombine(leftTextFuture, (leftTalk, leftText) -> {              PhonePlan plan = new PhonePlan();              plan.setLeftTalk(leftTalk);              plan.setLeftText(leftText);              return plan;          }).thenCombine(leftDataFuture, (plan, leftData) -> {              plan.setLeftData(leftData);              return plan;          });        return planFuture;  }  

     我們發現雖然異步編程可以獲得性能上的提升，但是編碼複雜度卻提升了很多，並且如果異步調用鏈太長，還容易導致回調地獄。

     ES2017 在編程語言級別提供了async/await關鍵字用於簡化異步編程，讓開發者以同步的方式編寫異步代碼，例如：

const leftTalk = await readLeftTalkPromise(phoneNo);  const leftText = await readLeftTextPromise(phoneNo);  const leftData = await readLeftDataPromise(phoneNo);  const phonePlan = new PhonePlan(leftTalk, leftText, leftData);   

     在Scala中使用 for 語句也可以簡化異步編程，例如：

for {    leftTalk <- leftTalkFuture    leftText <- leftTextFuture    leftData <- leftDataFuture  } yield new PhonePlan(leftTalk, leftText, leftData)  

     看到在其它語言中異步編程如此簡單，是不是很羨慕？ 別急， 在下一篇文章中，我們將會看到如何利用反應式編程簡化異步調用問題。

3 總結

       本文通過兩部分內容為大家介紹了反應式的基本概念。第一部分介紹什麼是反應式，包括反應式的發展歷史和一些相關項目。第二部分介紹為什麼要反應式，通過一個傳統的編程示例向大家闡述同步編程所面臨的問題和挑戰，尤其在微服務場景下，面對成千上萬的微服務接口以錯綜複雜的調用鏈，為了規避可能導致的雪崩風險，我們不得不對已有的架構進行無意義改造，不僅增加開發成本，而且導致部署和運維難度增加，同步編程方式已經深刻地影響到了我們的架構。但是不管怎麼說，反應式改造是一個長期的過程， 在這個過程中，我們需要不斷地完善基礎設施，同時也要注重對開發人員的培養， 因為反應式編程是對傳統方式的一次變革，編程模式和思維都需要進行轉換，這對於開發人員來說同樣是一次挑戰。轉型雖然痛苦，但是成功蛻變之後便會迎來新生。

4 參考

• 全面異步化：淘寶反應式架構升級探索
• 反應式宣言
• 反應式設計模式 Roland Kuhn, Brian Hanafee, Jamie Allen; 何品,邱嘉和,王石沖譯; 林煒翔校
• 反應式Web應用開發 Manuel Bernhardt; 張衛濱譯
• Reactive programming vs. Reactive systems，Jonas Bonér and Viktor Klang
• PayPal Blows Past 1 Billion Transactions Per Day Using Just 8 VMs With Akka, Scala, Kafka and Akka Streams