響應式架構與 RxJava 在有贊零售的實踐

隨着有贊零售業務的快速發展，系統和業務複雜度也在不斷提高。如何解決系統服務化後，多個系統之間的耦合，提高業務的響應時間與吞吐量，有效保證系統的健壯性和穩定性，是咱們面臨的主要問題。結合目前技術體系和業務特色的思考，咱們在業務中實踐了響應式架構以及RxJava框架，來解決系統與業務複雜所帶來的問題。

##實踐響應式架構 響應式架構是指業務組件和功能由事件驅動，每一個組件異步驅動，能夠並行和分佈式部署及運行。

響應式架構能夠帶來如下優點：

• 大幅度下降應用程序內部的耦合性
• 事件傳遞形式簡化了並行程序的開發工做，使開發人員無須與併發編程基礎元素打交道，同時能夠解決許多併發編程難題，如死鎖等。
• 響應式架構可以大幅度提升調用方法的安全性和速度。
• 對複雜業務系統的領域建模，響應式架構能夠自然支持。每一個系統組件就能夠對應到一個業務實體，業務實體之間經過接收事件來完成一次業務操做。

咱們使用響應式架構主要是爲解決多個系統間的屢次遠程調用帶來的分佈式問題，尤爲在長任務場景中，響應式架構顯得尤爲必要。

有贊連鎖出現後，隨着連鎖商家經營規模的擴張，會在系統中建立新的門店。建立新門店會引起一系列業務初始化工做，例如店鋪、員工、倉庫、商品、庫存等業務域，而且各業務域之間存在必定的依賴關係（如圖1所示），例如商品依賴倉庫初始化完成。

圖1 連鎖新建分店系統依賴關係

商家新增門店時，在店鋪初始化完成後，連鎖系統發送店鋪初始化成功消息，相應系統對事件進行響應，處理完成（成功／失敗）後將回執給連鎖系統，連鎖系統根據相關業務的反饋，決定是繼續通知下游業務，仍是結束整個過程。新建門店部分流程如圖2所示。

在建立門店業務中，每一個系統響應連鎖系統發出的消息，處理完成後進行回執。經過這種模式，業務系統自己不關心其餘系統是否成功或失敗，只需對通知的事件進行處理，總體初始化進度與異常處理由連鎖系統來控制。這種設計使得各業務系統之間沒有直接耦合並保持相互獨立。

圖2 連鎖體系新增分店消息驅動圖

上面的案例介紹了在複雜業務場景下系統間對響應式架構的實踐，系統內部一樣會遇到複雜業務場景。下面介紹下在系統內部應對複雜業務的實踐。

##RxJava在有贊零售實踐

Rxjava是用來編寫異步和基於消息的程序的類庫。RxJava在Android有着普遍的使用，主要應用在用戶界面繪製與服務端通信等場景。RxJava的核心思想是響應式編程以及事件、異步這兩個特色。響應式編程是一種經過異步和事件流來構建程序的編程模型。在複雜的業務開發中，最棘手的問題就是如何清晰直觀的展示覆雜的業務邏輯，而且方便後續的業務維護與擴展。

###響應式編程使得複雜業務邏輯更清晰

有贊零售的業務場景中有着複雜的業務邏輯，有贊目前提供多種產品供商家選擇，商家在不一樣產品進行切換時，爲了商家更好的體驗，不一樣業務的切換會進行數據初始化與處理。例若有贊微商城轉換到有贊零售。

這裏拿着微商城升級零售的業務場景給你們舉例。微商城升級爲零售時須要對商品進行轉換。首先初始化店鋪基礎信息。而後讀取商品流，將微商城的商品類型轉換成零售支持的商品類型。最後讀取規格，爲規格建立供應鏈商品庫，建立門店商品與添加網店商品的供應鏈商品關聯關係。總體轉換流程如圖3所示。圖中也畫出了能夠併發處理的場景。

圖3 微商城升級有贊零售流程

若是單純使用設計模式來解決上面這種場景單1、但業務邏輯特別複雜的場景，是很難作到的。也能夠看到除了初始化信息那一步，後面的商品模型轉化自始至終在業務中流轉的事件都是商品，這裏就可使用RxJava來優化業務代碼使得處理流程能夠併發，加快升級速度。

最終咱們按照圖3的流程處理升級邏輯，其中的併發場景，好比保存完零售商品後，併發處理庫存、和銷售渠道，使用rxjava封裝的方法幫助咱們進行併發操做。以下所示代碼結構清晰，對外屏蔽了複雜的併發處理邏輯。

Observable.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-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實現異步化和響應式編程。對於響應式架構的思想，咱們也在探索階段，並在部分業務場景進行實踐。將來面對愈來愈複雜的零售業務場景，會用響應式架構全面實現系統業務的異步化。總的來講響應式架構思想爲提高複雜業務系統健壯性、靈活性提供了強有力的支撐。後面你們若是想更多的討論響應式架構與編程的實踐，歡迎聯繫咱們。