https://mp.weixin.qq.com/s/7-h2w_iXrM5861iGTpftNQ

引言

響應式編程最簡單的定義是Reactive programming is programming with asynchronous data streams。無論是從Spring5中引入的響應式編程框架還是java9中集成的響應式流，都能看到響應式編程的影子?？梢哉f響應式編程代表了未來編程的方向。

響應式編程其天然就是非阻塞的，當數據準備完成后自動觸發下一個動作而不是等待數據完成。這種思想再結合異步化編程使得我們在統一線程模型，降低多線程編程成本的同時提升整個系統的吞吐量。

閑魚當前的業務場景本身足夠復雜，然而目前大規模使用的仍然是阻塞式編程，每個業務場景的流程被串行同步執行。它最大的問題在于大量線程等待IO(日志記錄，RPC調用，HTTP請求等)完成。有鑒于此，閑魚引入RxJava 2.0來作為響應式編程框架對應用進行異步化改造。目前app內的魚塘首頁，留言列表以及魚塘詳情已經完成異步化改造并上線。

響應式異步化編程實踐

異步化編程不同于傳統的阻塞式編程，他們的區別可以用下面的圖來表達，C代表computation任務，I代表IO任務。原來的一個線程中執行完所有的computation和IO任務轉變成現在computation和IO任務分離，這樣做的好處是CPU只關注computation任務，理論上CPU核數個線程就能滿足所有的computation任務，大大減少線程切換開銷。

線程模型 RxJava框架背后是線程模型的演進。在引入RxJava前傳統Java自帶多線程框架(Executor)和異步Future,它們的問題在于

• 線程池無法統一。開發在自己的場景下都可能去定義線程池。

• 上下文切換。隨著線程數的增多，線程上下文的切換必然增多。

• Future異步方式仍然是一種阻塞等待式方法。

RxJava結合全異步編程方式的優勢在于

• 線程利用率大幅提高。當線程需要做阻塞操作時及時切換避免長時間占有線程，整個流程無阻塞。

• RxJava統一了線程池模型。我們可以根據不同的場景選擇對應的線程調度模型。

• 極致線程模型變得可能。理論上只需要cpu核數個線程就可以運行所有computation任務，這意味線程切換帶來的開銷幾乎被消除。

應用改造

響應式全異步編程天生拒絕阻塞，任何阻塞點都可能導致性能的退步。更嚴重的如果我們控制了線程數，當任務因為阻塞而產生堆積，隨著堆積的任務變多應用會gc影響線上服務。目前閑魚應用中存在4類阻塞點

• 應用日志。

• HTTP請求。

• RPC調用。

• 緩存讀寫。

閑魚中的日志不僅僅作為異常輸出手段，也是數據統計的一個主要方法。因為日志的性能顯得很重要。雖然log4j和logback都提供了異步方式，但是它們本質上還是基于鎖來實現。log4j2是新一代的基于LMAX Disruptor的無鎖異步日志系統，在多線程程序中，其吞吐量比log4j和logback高10倍左右。

而HTTP，RPC以及緩存的讀寫都需要改造成純異步方式：當請求發生時線程被釋放，請求完成后繼續在新的線程中執行余下業務流程。

執行范式

對閑魚中的場景進行一下歸納，可以梳理出三種執行范式，下面會分別簡單展示這三種范式。為了簡單起見，其中涉及到的IO操作都用一個函數加以抽象。

串行請求?以常用的電商場景為例，查看我買的商品詳情。業務流程上要先查詢訂單詳情，然后從訂單詳情中拿到商品，最后根據商品去查詢商品詳情。

//查訂單

Flowable<Order> orderFlow = Flowable.fromCallable(() -> queryOrder(orderId));

//查商品

Flowable<Item> itemFlow = orderFlow.flatMap(order -> Flowable.fromCallable(() -> queryItem(order.getId())));

并發請求?商品詳情中往往還會有一些額外的信息，比如瀏覽量以及留言內容。瀏覽量和留言它們彼此是互不依賴的，但是都依賴查到的商品信息。

Flowable<Detail> detailFlow = itemFlow.flatMap(item -> {

? ?//瀏覽量

? ?Flowable<Long> pvFlow = Flowable.fromCallable(() -> queryItemPv(item.getId()));

? ?//留言

? ?Flowable<Comment> commentFlow = Flowable.fromCallable(() -> queryItemComment(item.getId()));

? ?//瀏覽，留言，商品共同組成detail

? ?return Flowable.zip(pvFlow,commentFlow,(pv,comment) -> buildItemDetail(item,pv,comment));

});

更新緩存?熱門商品詳情往往意味高并發訪問，我們可以將這些數據緩存來減輕數據庫的壓力。但是緩存成功與否都不影響本次請求(緩存失敗導致下次請求仍然走db，本次請求的數據仍然返回給用戶)。更新緩存本質上代表這樣一類操作，它們在請求完成后執行一些額外的操作(緩存，通知用戶，日志記錄等)，這些額外操作的成功與否不對主流程造成任何影響。

//更新緩存，doOnNext不會對流的結果造成任何影響，只是觸發一個操作

detailFlow.doOnNext(detail -> cache(detail));

需要強調的是上面流中所有的方法都對應著一次IO操作(RPC，緩存讀寫等)，這些IO操作都應該是全異步方式實現(不能等待IO完成，而是IO完成后主動喚醒)。幸運的是已經有一些第三方庫來幫我們完成這些IO異步化。

改造結果

我們對改造完的接口進行了一輪性能測試，分別從接口rt，線程數以及cpu利用率三個方面對阻塞式執行和響應式純異步方式進行對比

rt?因為我們在異步方式中增加了并行操作， 所以rt降低是必然的，rt下降50%左右。當請求QPS達到650的時候，傳統阻塞方式rt飆升，服務開始不可用；響應式純異步方式rt較為穩定，QPS達到850的時候開始明顯上漲。

線程數?阻塞式執行方式因為在IO發生時線程會等待IO完成，而異步方式下線程直接釋放，所以異步方式下線程利用效率明顯更高。下面的測試結果也表面異步方式的線程數一直較為穩定；阻塞模式下線程在QPS到達650時被耗盡，這意味著新的請求將被直接拒絕。

CPU?由于異步模式下我們增加了并發，因此CPU使用必然會較阻塞模式高，測試結果也說明了這一點。然而當QPS到達650的時候，阻塞模式下服務已經不可用，因此CPU利用率最高只能到達75%左右；而異步模式CPU能夠到達97%。

總結

本文介紹了響應式編程在閑魚的應用現狀，我們選用了RxJava作為響應式編程框架，并選取了閑魚留言列表，魚塘首頁以及魚塘詳情頁進行改造并進行了性能測試。測試數據表明閑魚群聊首頁場景下，在rt降低50%的同時整個系統的吞吐量能提升約30%。

目前響應式純異步編程在閑魚仍然處于起步階段，我們的線程池模型仍然沒法做到極致(暫時沒法完全消除阻塞點)。接下來會朝著極致線程模型進行嘗試，可能的改造點包括

• 阻塞點消除。

• cpu核數個線程來調度所有的computation任務。

• 底層的IO線程模型統一?，F在IO操作(RPC,HTTP,緩存)都是各自維護自己的線程池，理論上這些線程池都可以趨于統一。

轉載于:https://www.cnblogs.com/davidwang456/articles/10175728.html

總結

以上是生活随笔為你收集整理的RxJava在闲鱼系统吞吐量提升上的实践的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。