2. 程式人生 > >【轉】聊聊java高併發系統之非同步非阻塞

【轉】聊聊java高併發系統之非同步非阻塞

阿新 • • 發佈：2019-02-01

在做電商系統時，流量入口如首頁、活動頁、商品詳情頁等系統承載了網站的大部分流量，而這些系統的主要職責包括聚合資料拼裝模板、熱點統計、快取、下游功能降級開關、託底資料等等。其中聚合資料需要呼叫其它多個系統服務獲取資料、拼裝資料/模板然後返回給前端，聚合資料來源主要有依賴系統/服務、快取、資料庫等;而系統之間的呼叫可以通過如http介面呼叫(如HttpClient)、SOA服務呼叫(如dubbo、thrift)等等。

在Java中，如使用Tomcat，一個請求會分配一個執行緒進行請求處理，該執行緒負責獲取資料、拼裝資料或模板然後返回給前端;在同步呼叫獲取資料介面的情況下(等待依賴系統返回資料)，整個執行緒是一直被佔用並阻塞的。如果有大量的這種請求，每個請求佔用一個執行緒，但執行緒一直處於阻塞，降低了系統的吞吐量，這將導致應用的吞吐量下降;我們希望在呼叫依賴的服務響應比較慢，此時應該讓出執行緒和CPU來處理下一個請求，當依賴的服務返回了再分配相應的執行緒來繼續處理。而這應該有更好的解決方案：非同步/協程。而Java是不支援協程的(雖然有些Java框架說支援，但還是高層API的封裝)，因此在Java中我們還可以使用非同步來提升吞吐量。目前java一些開源框架(HttpClient\HttpAsyncClient、dubbo、thrift等等)大部分都支援。

幾種呼叫方式

同步阻塞呼叫

即序列呼叫，響應時間為所有服務的響應時間總和;

半非同步(非同步Future)

執行緒池，非同步Future，使用場景：併發請求多服務，總耗時為最長響應時間;提升總響應時間，但是阻塞主請求執行緒，高併發時依然會造成執行緒數過多，CPU上下文切換;

全非同步(Callback)

Callback方式呼叫，使用場景：不考慮回撥時間且只能對結果做簡單處理，如果依賴服務是兩個或兩個以上服務，則不能合併兩個服務的處理結果;不阻塞主請求執行緒，但使用場景有限。

非同步回撥鏈式編排

非同步回撥鏈式編排(JDK8 CompletableFuture)，使用場景：其實不是非同步呼叫方式，只是對依賴多服務的Callback呼叫結果處理做結果編排，來彌補Callback的不足，從而實現全非同步鏈式呼叫。

接下來看看如何設計利用全非同步Callback呼叫和非同步回撥鏈式編排處理結果來實現全非同步系統設計。

同步阻塞呼叫

  1. public class Test { 
  2.    public static void main(String[] args) throws Exception { 
  3.        RpcService rpcService = new RpcService(); 
  4.        HttpService httpService = new HttpService(); 
  5.        //耗時10ms 
  6.        Map<String, String>result1 = rpcService
    .getRpcResult(); 
  7.        //耗時20ms 
  8.        Integer result2 = httpService.getHttpResult(); 
  9.        //總耗時30ms 
  10.     } 
  11.    static class RpcService { 
  12.        Map<String, String> getRpcResult() throws Exception { 
  13.            //呼叫遠端方法（遠端方法耗時約10ms，可以使用Thread.sleep模擬） 
  14.        } 
  15.     } 
  16.    static class HttpService { 
  17.        Integer getHttpResult() throws Exception { 
  18.            //呼叫遠端方法（遠端方法耗時約20ms，可以使用Thread.sleep模擬） 
  19.            Thread.sleep(20); 
  20.            return 0; 
  21.        } 
  22.     } 
  23. } 

半非同步(非同步Future)

  1. public class Test { 
  2.    final static ExecutorService executor = Executors.newFixedThreadPool(2); 
  3.    public static void main(String[] args) { 
  4.        RpcService rpcService = new RpcService(); 
  5.        HttpService httpService = new HttpService(); 
  6.        Future<Map<String, String>>future1 = null; 
  7.        Future<Integer>future2 = null; 
  8.        try { 
  9.            future1 = executor.submit(() -> rpcService.getRpcResult()); 
  10.            future2 = executor.submit(() -> httpService.getHttpResult()); 
  11.            //耗時10ms 
  12.            Map<String, String>result1 = future1.get(300, TimeUnit.MILLISECONDS); 
  13.            //耗時20ms 
  14.            Integer result2 = future2.get(300, TimeUnit.MILLISECONDS); 
  15.            //總耗時20ms 
  16.        } catch (Exception e) { 
  17.            if (future1 != null) { 
  18.                 future1.cancel(true); 
  19.            } 
  20.            if (future2 != null) { 
  21.                 future2.cancel(true); 
  22.            } 
  23.            throw new RuntimeException(e); 
  24.        } 
  25.     } 
  26.    static class RpcService { 
  27.        Map<String, String> getRpcResult() throws Exception { 
  28.            //呼叫遠端方法（遠端方法耗時約10ms，可以使用Thread.sleep模擬） 
  29.        } 
  30.     } 
  31.    static class HttpService { 
  32.        Integer getHttpResult() throws Exception { 
  33.            //呼叫遠端方法（遠端方法耗時約20ms，可以使用Thread.sleep模擬） 
  34.        } 
  35.     } 
  36. } 

全非同步(Callback)

  1. public class AsyncTest { 
  2. public staticHttpAsyncClient httpAsyncClient; 
  3.    public static CompletableFuture<String> getHttpData(String url) { 
  4.        CompletableFuture asyncFuture = new CompletableFuture(); 
  5.        HttpPost post = new HttpPost(url); 
  6.        HttpAsyncRequestProducer producer = HttpAsyncMethods.create(post); 
  7.        AsyncCharConsumer<HttpResponse>consumer = newAsyncCharConsumer<HttpResponse>() { 
  8.             HttpResponse response; 
  9.            protected HttpResponse buildResult(final HttpContext context) { 
  10.                 return response; 
  11.            } 
  12. …... 
  13.        }; 
  14.        FutureCallback callback = new FutureCallback<HttpResponse>() { 
  15.            public void completed(HttpResponse response) { 
  16.                asyncFuture.complete(EntityUtils.toString(response.getEntity())); 
  17.            } 
  18. …... 
  19.        }; 
  20.        httpAsyncClient.execute(producer, consumer, callback); 
  21.        return asyncFuture; 
  22.     } 
  23.    public static void main(String[] args) throws Exception { 
  24.        AsyncTest.getHttpData("http://www.jd.com"); 
  25.        Thread.sleep(1000000); 
  26.     } 
  27. } 

本示例使用HttpAsyncClient演示。

非同步回撥鏈式編排

CompletableFuture提供了50多個API，可以滿足所需的各種場景的非同步處理的編排，在此列舉三個場景：

場景1：三個服務併發非同步呼叫，返回CompletableFuture，不阻塞主執行緒;

方法test1：

  1. public static void test1() throws Exception { 
  2.       HelloClientDemoTest service = new HelloClientDemoTest(); 
  3.       /** 
  4.        * 場景1 兩個以上服務併發非同步呼叫，返回CompletableFuture,不阻塞主執行緒 
  5.        * 並且兩個服務也是非同步非阻塞呼叫 
  6.        */ 
  7.       CompletableFuture future1 = service.getHttpData("http://www.jd.com"); 
  8.       CompletableFuture future2 = service.getHttpData("http://www.jd.com"); 
  9.       CompletableFuture future3 =service.getHttpData("http://www.jd.com"); 
  10.       List<CompletableFuture>futureList = Lists.newArrayList(future1,future2, future3); 
  11.       CompletableFuture<Void>allDoneFuture =CompletableFuture.allOf(futureList.toArray(newCompletableFuture[futureList.size()])); 
  12.       CompletableFuture<String>future4 =allDoneFuture.thenApply(v -> { 
  13.            List<Object>result =futureList.stream().map(CompletableFuture::join) 
  14.                   .collect(Collectors.toList()); 
  15.            //注意順序 
  16.            String result1 = (String)result.get(0); 
  17.            String result2 = (String)result.get(1); 
  18.            String result3 = (String)result.get(2); 
  19.            //處理業務.... 
  20.            return result1 + result2 + result3; 
  21.        }).exceptionally(e -> { 
  22.            //e.printStackTrace(); 
  23.            return ""; 
  24.        }); 
  25.       //返回 
  26.    } 

場景2、兩個服務併發非同步呼叫，返回CompletableFuture，不阻塞主執行緒;

方法test2：

  1. public void test2() throws Exception { 
  2.       HelloClientDemoTest service = new HelloClientDemoTest(); 
  3.       /** 
  4.        * 場景2 兩個介面併發非同步呼叫，返回CompletableFuture,不阻塞主執行緒 
  5.        * 並且兩個服務也是非同步非阻塞呼叫 
  6.        */ 
  7.       CompletableFuture future1 = service.getHttpData("http://www.jd.com"); 
  8.       CompletableFuture future2 =service.getHttpData("http://www.jd.com"); 
  9.       CompletableFuture future3 =future1.thenCombine(future2, (f1, f2) -> { 
  10.            //處理業務.... 
  11.            return f1 + "," + f2; 
  12.        }).exceptionally(e -> { 
  13.            return ""; 
  14.        }); 
  15.       //返回 
  16.    } 

場景3、兩個服務，併發非同步呼叫兩個服務，並且一個服務的結果返回後再次呼叫另一服務，然後將三個結果後並處理，返回CompletableFuture，整個處理過程中不阻塞任何執行緒;

方法test3：

  1. publicvoid test3() throws Exception { 
  2.        HelloClientDemoTest service = new HelloClientDemoTest(); 
  3.        /** 
  4.         * 場景3 兩請求依賴呼叫，然後與另一服務結果組合處理，返回CompletableFuture,不阻塞主執行緒 
  5.         * 並且兩個服務也是非同步非阻塞呼叫 
  6.         */ 
  7.         CompletableFuture future1 = service.getHttpData("http://www.jd.com"); 
  8.         CompletableFuture future2 = service.getHttpData("http://www.jd.com"); 
  9.         CompletableFuture<String>future3= future1.thenApply((param) -> { 
  10.             CompletableFuture future4 =service.getHttpData("http://www.jd.com"); 
  11.             return future4; 
  12.         }); 
  13.         CompletableFuture future5 =future2.thenCombine(future3, (f2, f3) -> { 
  14.             //....處理業務 
  15.             return f2 + "," + f3; 
  16.         }).exceptionally(e -> { 
  17.             return ""; 
  18.         }); 
  19.         //返回future5 
  20.     } 

全非同步Web系統設計

主要技術：servlet3，JDK8 CompletableFuture，支援非同步Callback呼叫的RPC框架。

先看一下處理流程圖：

servlet3：Servlet 接收到請求之後，可能首先需要對請求攜帶的資料進行一些預處理;接著，Servlet 執行緒將請求轉交給一個非同步執行緒來執行業務處理，執行緒本身返回至容器。針對業務處理較耗時的情況，這將大大減少伺服器資源的佔用，並且提高併發處理速度。servlet3可參考商品詳情頁系統的Servlet3非同步化實踐，結合其中講解的servlet3整合：

  1. public void submitFuture(finalHttpServletRequest req, final Callable<CompletableFuture> task) throwsException{ 
  2.        final String uri = req.getRequestURI(); 
  3.        final Map<String, String[]>params = req.getParameterMap(); 
  4.        final AsyncContext asyncContext = req.startAsync(); 
  5.        asyncContext.getRequest().setAttribute("uri", uri); 
  6.        asyncContext.getRequest().setAttribute("params", params); 
  7.        asyncContext.setTimeout(asyncTimeoutInSeconds * 1000); 
  8.        if(asyncListener != null) { 
  9.            asyncContext.addListener(asyncListener); 
  10.        } 
  11.        CompletableFuture future = task.call(); 
  12.        future.thenAccept(result -> { 
  13.            HttpServletResponse resp = (HttpServletResponse)asyncContext.getResponse(); 
  14.            try { 
  15.                 if(result instanceof String) { 
  16.                     byte[] bytes = new byte[0]; 
  17.                     if (StringUtils.isBlank(result)){ 
  18.                        resp.setContentType("text/html;charset=gbk"); 
  19.                        resp.setContentLength(0); 
  20.                     } else { 
  21.                         bytes =result.getBytes("GBK"); 
  22.                     } 
  23.                    //resp.setBufferSize(bytes.length); 
  24.                    resp.setContentType("text/html;charset=gbk"); 
  25.                    if(StringUtils.isNotBlank(localIp)) { 
  26.                        resp.setHeader("t.ser", localIp); 
  27.                     } 
  28.                    resp.setContentLength(bytes.length); 
  29.                    resp.getOutputStream().write(bytes); 
  30.                 } else { 
  31.                     write(resp,JSONUtils.toJSON(result)); 
  32.                 } 
  33.            } catch (Throwable e) { 
  34.                resp.setStatus(HttpServletResponse.SC_INTERNAL_SERVER_ERROR); //程式內部錯誤 
  35.                 try { 
  36.                     LOG.error("get infoerror, uri : {},  params : {}", uri,JSONUtils.toJSON(params), e); 
  37.                 } catch (Exception ex) { 
  38.                 } 
  39.            } finally { 
  40.                 asyncContext.complete(); 
  41.            } 
  42.        }).exceptionally(e -> { 
  43.            asyncContext.complete(); 
  44.            return null; 
  45.        }); 
  46. } 

另外還有Java中協程庫Quasar，可參考《Java的纖程庫 - Quasar》，目前沒有在應用中使用並在測試FiberHttpServlet的時候遇到很多坑，日後把Quasar自如運用後形成日記，希望能結實更多的朋友一起研究，踩坑。

作者：孫偉，目前負責京東商品詳情頁統一服務系統，寫過java，寫過ngx_lua，還寫過storm等，喜歡學習研究新事物。

【本文來自51CTO專欄作者張開濤的微信公眾號(開濤的部落格)，公眾號id: kaitao-1234567】

相關推薦

【轉】聊聊java高併發系統之非同步非阻塞

在做電商系統時，流量入口如首頁、活動頁、商品詳情頁等系統承載了網站的大部分流量，而這些系統的主要職責包括聚合資料拼裝模板、熱點統計、快取、下游功能降級開關、託底資料等等。其中聚合資料需要呼叫其它多個系統服務獲取資料、拼裝資料/模板然後返回給前端，聚合資料來源主要有依賴系統/服務、快取、資料庫等;而系統之間

聊聊java高併發系統之非同步非阻塞

幾種呼叫方式 同步阻塞呼叫 即序列呼叫，響應時間為所有服務的響應時間總和； 半非同步(非同步Future) 執行緒池，非同步Future，使用場景：併發請求多服務，總耗時為最長響應時間；提升總響應時間，但是阻塞主請求執行緒，高併發時依然會造成執行緒數過多，CPU上下文切換； 全非同步(Cal

java高併發系統之非同步非阻塞

在做電商系統時，流量入口如首頁、活動頁、商品詳情頁等系統承載了網站的大部分流量，而這些系統的主要職責包括聚合資料拼裝模板、熱點統計、快取、下游功能降級開關、託底資料等等。其中聚合資料需要呼叫其它多個系統服務獲取資料、拼裝資料/模板然後返回給前端，聚合資料來源主要有依賴系統

聊聊高併發系統之降級特技（轉）

在開發高併發系統時有三把利器用來保護系統：快取、降級和限流。之前已經有一些文章介紹過快取和限流了。本文將詳細聊聊降級。當訪問量劇增、服務出現問題（如響應時間慢或不響應）或非核心服務影響到核心流程的效能時，仍然需要保證服務還是可用的，即使是有損服務。系統可以根據一些關鍵資料進行自動降級，也

聊聊高併發系統之限流特技（二）(轉)

上一篇《聊聊高併發系統限流特技-1》講了限流演算法、應用級限流、分散式限流；本篇將介紹接入層限流實現。 接入層限流 接入層通常指請求流量的入口，該層的主要目的有：負載均衡、非法請求過濾、請求聚合、快取、降級、限流、 A/B 測試、服務質量監控

【轉】VMware Converter遷移linux系統虛擬機

失敗 automatic 測試 tac 精簡 基本 init 安裝 image 原始出處 今天接到一個需求，遷移一臺linux的業務系統到vCenter雲管理平臺，其中遇到一些問題，於是進行了排錯，這個過程與大家分享，下面把整個步驟進行截圖說明。 1. 首先，登錄到VM

【轉】Win7 + Ubuntu16.04 雙系統安裝

hdd 免費 發現 bottom 處理 數據 掛載 練習 b- 【尊重原創】 http://blog.csdn.net/u012260238/article/details/52713546 整個安裝過程主要分為以下幾步： 一、 下載ubuntu16.04鏡像;

【轉】給Java說句公道話

paper 計算 復雜 解決問題 畢業 doc 就是 限制 sonar 有些人問我，在現有的語言裏面，有什麽好的推薦？我說：“Java。” 他們很驚訝：“什麽？Java！” 所以我現在來解釋一下。 Java超越了所有咒罵它的“動態語言” 也許是因為年輕人的逆反心理，人們都不

【轉】關於java 單元測試Junit4和Mock的一些總結

原文出處請點選這裡 1. 單元測試的必要性 最近專案有在寫java程式碼的單元測試，然後在思考一個問題，為什麼要寫單元測試？？單元測試寫了有什麼用？？百度了一圈，如下： 軟體質量最簡單、最有效的保證； 是目的碼最清晰、最有效的文件；

【轉】 實時Android語音對講系統架構

本文屬於Android區域網內的語音對講專案系列，《通過UDP廣播實現Android區域網Peer Discovering》實現了局域網內的廣播及多播通訊，本文將重點說明系統架構，音訊訊號的實時錄製、播放及編解碼相關技術。 本文主要包含以下內容： AudioRecord、AudioTr

【轉】用Java實現網路語音訊號傳送

本文轉載自部落格：https://www.aliyun.com/jiaocheng/347518.html ----------------------------------------------------------------------------------------------

【轉】杜絕JAVA軟體開發的五大誤區

越來越多人開始使用Java，但是他們大多數人沒有做好足夠的思想準備(沒有接受OO思想體系相關培訓)，以致不能很好駕馭Java專案，甚至導致開發後的Java系統性能緩慢甚至經常當機。很多人覺得這是Java複雜導致，其實根本原因在於:我們原先掌握的關於軟體知識(OO方面)

【轉】Elasticsearch Java Rest Client 指南

sta eve oat bubuko 增刪改查 cli for 在一起 min 原文地址：https://www.jianshu.com/p/d2c8326e8fa3 仔細看了下，5.6版本的RestHighLevelClient就這麽些API，有興趣的朋友可以去看

聊聊高併發系統之HTTP快取

簡介 最近遇到很多人來諮詢我關於瀏覽器快取的一些問題，而這些問題都是類似的，因此總結本文來解答以後遇到類似問題的朋友。 因本文主要以瀏覽器快取場景介紹，所以非瀏覽器場景下的一些用法本文不會介紹，而且本文以chrome為測試瀏覽器。 瀏覽器快取是指當我們使用瀏覽器訪問一些網站頁面或者http服務時，根

聊聊高併發系統之佇列術

佇列在資料結構中是一種線性表，從一端插入資料，然後從另一端刪除資料。本文目的不是講解各種佇列演算法，而是在應用層面講述使用佇列能解決哪些場景問題。 在我開發過的系統中，不是所有的業務都必須實時處理、不是所有的請求都必須實時反饋結果給使用者、不是所有的請求/處理都必須100%處理成功、不知道誰依賴“

Java高併發系統的限流策略

概要在大資料量高併發訪問時，經常會出現服務或介面面對暴漲的請求而不可用的情況，甚至引發連鎖反映導致整個系統崩潰。此時你需要使用的技術手段之一就是限流，當請求達到一定的併發數或速率，就進行等待、排隊、降級、拒絕服務等。在開發高併發系統時有三把利器用來保護系統：快取、降級和限流。快取快取比較好理解，在大型高併發系

【轉】Windows和Ubuntu雙系統，修復UEFI引導的兩種辦法

讀者在讀過我的安裝Ubuntu和Windows 8雙系統教程以後(見http://www.linuxidc.com/Linux/2015-01/111796.htm)，碰到的主要的問題是電腦直接啟動到Windows 8而沒有出現啟動Ubuntu的選項。 這裡有兩種修復EFI啟

【Golang】百萬級高併發實踐

轉自：https://blog.csdn.net/jeanphorn/article/details/79018205寫在前面Go語言作為新興的語言，最近發展勢頭很是迅猛，其最大的特點就是原生支援併發。它使用的是“協程（goroutine）模型”，和傳統基於 OS 執行緒和程

聊聊高併發系統之限流特技

在開發高併發系統時有三把利器用來保護系統：快取、降級和限流。快取的目的是提升系統訪問速度和增大系統能處理的容量，可謂是抗高併發流量的銀彈；而降級是當服務出問題或者影響到核心流程的效能則需要暫時遮蔽掉，待高峰或者問題解決後再開啟；而有些場景並不能用快取和降級來解決，比如

【轉】深入理解Linux的系統呼叫

　　一、 什麼是系統呼叫 　　 　　在Linux的世界裡，我們經常會遇到系統呼叫這一術語，所謂系統呼叫，就是核心提供的、功能十分強大的一系列的函式。這些系統呼叫是在核心中實現的，再通過一定的方式把系統呼叫給使用者，一般都通過門(gate)陷入(trap)實現。系統呼叫是使用