響應式架構與 RxJava 在有贊零售的實踐
文章概要
隨著有贊零售業務的快速發展，系統和業務複雜度也在不斷提升。如何解決系統服務化後，多個系統之間的耦合，提升業務的響應時間與吞吐量，有效保證系統的健壯性和穩定性，是我們面臨的主要問題。結合目前技術體系和業務特點的思考，我們在業務中實踐了響應式架構以及 RxJava 框架，來解決系統與業務複雜所帶來的問題。
一、實踐響應式架構
響應式架構是指業務元件和功能由事件驅動，每個元件非同步驅動，可以並行和分散式部署及執行。
響應式架構可以帶來以下優勢：
大幅度降低應用程式內部的耦合性
事件傳遞形式簡化了並行程式的開發工作，使開發人員無須與併發程式設計基礎元素打交道，同時可以解決許多併發程式設計難題，如死鎖等。
響應式架構能夠大幅度提高呼叫方法的安全性和速度。
對複雜業務系統的領域建模，響應式架構可以天然支援。每個系統元件就可以對應到一個業務實體，業務實體之間通過接收事件來完成一次業務操作。
我們使用響應式架構主要是為解決多個系統間的多次遠端呼叫帶來的分散式問題，尤其在長任務場景中，響應式架構顯得尤其必要。
有贊連鎖出現後，隨著連鎖商家經營規模的擴張，會在系統中建立新的門店。建立新門店會引發一系列業務初始化工作，例如店鋪、員工、倉庫、商品、庫存等業務域，並且各業務域之間存在一定的依賴關係（如圖1所示），例如商品依賴倉庫初始化完成。
圖1 連鎖新建分店系統依賴關係
商家新增門店時，在店鋪初始化完成後，連鎖系統傳送店鋪初始化成功訊息，相應系統對事件進行響應，處理完成（成功／失敗）後將回執給連鎖系統，連鎖系統根據相關業務的反饋，決定是繼續通知下游業務，還是結束整個過程。新建門店部分流程如圖2所示。
在建立門店業務中，每個系統響應連鎖系統發出的訊息，處理完成後進行回執。通過這種模式，業務系統本身不關心其他系統是否成功或失敗，只需對通知的事件進行處理，整體初始化進度與異常處理由連鎖系統來控制。這種設計使得各業務系統之間沒有直接耦合並保持相互獨立。
圖2 連鎖體系新增分店訊息驅動圖
上面的案例介紹了在複雜業務場景下系統間對響應式架構的實踐，系統內部同樣會遇到複雜業務場景。下面介紹下在系統內部應對複雜業務的實踐。
二、RxJava在有贊零售實踐
Rxjava 是用來編寫非同步和基於訊息的程式的類庫。RxJava 在 Android 有著廣泛的使用，主要應用在使用者介面繪製與服務端通訊等場景。RxJava 的核心思想是響應式程式設計以及事件、非同步這兩個特點。響應式程式設計是一種通過非同步和事件流來構建程式的程式設計模型。在複雜的業務開發中，最棘手的問題就是如何清晰直觀的展現複雜的業務邏輯，並且方便後續的業務維護與擴充套件。
2.1 響應式程式設計使得複雜業務邏輯更清晰
有贊零售的業務場景中有著複雜的業務邏輯，有贊目前提供多種產品供商家選擇，商家在不同產品進行切換時，為了商家更好的體驗，不同業務的切換會進行資料初始化與處理。例如有贊微商城轉換到有贊零售。
這裡拿著微商城升級零售的業務場景給大家舉例。微商城升級為零售時需要對商品進行轉換。首先初始化店鋪基礎資訊。然後讀取商品流，將微商城的商品型別轉換成零售支援的商品型別。最後讀取規格，為規格建立供應鏈商品庫，建立門店商品與新增網店商品的供應鏈商品關聯關係。整體轉換流程如圖3所示。圖中也畫出了可以併發處理的場景。
圖3 微商城升級有贊零售流程
如果單純使用設計模式來解決上面這種場景單一、但業務邏輯特別複雜的場景，是很難做到的。也可以看到除了初始化資訊那一步，後面的商品模型轉化自始至終在業務中流轉的事件都是商品，這裡就可以使用 RxJava 來優化業務程式碼使得處理流程可以併發，加快升級速度。
最終我們按照圖3的流程處理升級邏輯，其中的併發場景，比如儲存完零售商品後，併發處理庫存、和銷售渠道，使用 rxjava 封裝的方法幫助我們進行併發操作。如下所示程式碼結構清晰，對外遮蔽了複雜的併發處理邏輯。
.zip(
callAsync(()->處理庫存相關操作),
callAsync(()->更新商品庫門店銷售渠道),
callAsync(()->建立商品庫與網店商品關聯關係),
(sku1,sku2,sku3)-> sku
).blockingFirst();
最終我們的整體的程式碼：
UpgradeItem
.listItems(manager, shop)
.flatMap(item-> fromCallable(()->更新為零售商品型別))
.flatMap(item-> fromCallable(()->併發處理商品操作),
true
)
.flatMap(item-> 商品流轉化為sku流,
true
)
.flatMap(sku-> fromCallable(()->儲存零售商品))
.flatMap(sku-> fromCallable(()->併發處理儲存商品後續操作,
true
)
.subscribeOn(
Schedulers
.io());
整個商品處理流程就是上面這段程式碼，一目瞭然，後面擴充套件可以自己在中間加入處理流程，也可以在對應業務方法中修改邏輯。
2.2 多服務、資料來源組合
隨著微服務架構興起，我們將不同的業務域拆分成不同的系統。這樣方便了系統的維護，提升了系統的擴充套件性，但是給上層業務系統也帶來了很多麻煩。往往我們為了展示一個頁面會涉及到 2-3 個或更多的應用，而多次的分散式呼叫不但使得系統的rt增加，也使得核心頁面的出錯風險更高。
降低rt：在假設第三方介面已經達到效能頂點的情況下，併發是解決多次分散式呼叫降低rt的常用方法。
自動降級：傳統程式設計方法中，自動降級處理，意味著我們程式碼中會出現一大堆try/catch，而使用 rxjava，我們可以直接定義當流處理異常時，程式需要怎麼做，這樣的程式碼看起來非常簡潔。
商品搜尋作為商品管理的核心入口，根據不同場景聚合商品、優惠、庫存等資訊。由於商品列表頁展示的資訊涉及到多服務資料的整合，一方面需要保證整個介面的 rt，另一方面不希望由於一個商品資料或外部服務的異常影響到整個商品列表的載入。因此該場景非常適用於 RxJava。
最終我們的程式碼
1.根據入參獲取商品載入器
//只有包含的merger才會載入
List
<
SkuAttrMerger
> validMergers =
Observable
.fromIterable(skuAttrMergers).filter(loader -> request.getAttributes().contains(loader.supportAttribute().getValue())).toList().blockingGet();
2.根據 es 結果獲取商品各個屬性詳情並載入到 SkuAttrContext 中（某類屬性載入失敗則忽略）
//呼叫load併發載入資料到商品屬性上下文中
Observable
.fromIterable(商品資訊載入器列表)
.flatMap(商品資訊載入器->
Observable
.fromCallable(() ->非同步載入商品資訊))
.onErrorResumeNext(
Observable
.empty())
//如果失敗則忽略
.subscribeOn(
Schedulers
.io()),
false
,執行緒數(為載入器數
量)).blockingSubscribe();
3.組裝搜尋結果（如果某個 sku 組裝失敗則直接忽略）
//呼叫merge將資料合併到目標物件
商品搜尋返回結果列表 =
Observable
.fromIterable(商品id列表)
.map(商品id->初始化商品搜尋結果返回物件)
.flatMap(商品搜尋結果返回物件-> {
val observables=
Observable
.fromIterable(商品載入器列表)
.map(loader ->
Observable
.fromCallable(() ->合併每個sku的不同屬性)).toList().blockingGet();
return
Observable
.zipIterable(observables, (a) -> sku,
false
, 執行緒數)
.onErrorResumeNext(
Observable
.empty());
//如果失敗則忽略
},
false
,
1
)
.toList()
.blockingGet();
三、後記
本文主要介紹了響應式架構與 RxJava 在有贊零售的使用場景。目前我們對響應式架構的實踐方式是：在系統間使用訊息中介軟體來進行實現，在系統內則使用 RxJava 實現非同步化和響應式程式設計。對於響應式架構的思想，我們也在探索階段，並在部分業務場景進行實踐。未來面對越來越複雜的零售業務場景，會用響應式架構全面實現系統業務的非同步化。總的來說響應式架構思想為提升複雜業務系統健壯性、靈活性提供了強有力的支撐。後面大家如果想更多的討論響應式架構與程式設計的實踐，歡迎聯絡我們。
原文連結
https://mp.weixin.qq.com/s/HmE0oUq4escwlUYcgEByUg