CompletableFuture提供了50多個(gè)API，可以滿足所需的各種場(chǎng)景的異步處理的編排，在此列舉三個(gè)場(chǎng)景：

場(chǎng)景1：三個(gè)服務(wù)并發(fā)異步調(diào)用，返回CompletableFuture，不阻塞主線程;

方法test1：

public static void test1() throws Exception { 
      HelloClientDemoTest service = new HelloClientDemoTest(); 
      /** 
       * 場(chǎng)景1 兩個(gè)以上服務(wù)并發(fā)異步調(diào)用，返回CompletableFuture,不阻塞主線程 
       * 并且兩個(gè)服務(wù)也是異步非阻塞調(diào)用 
       */ 
      CompletableFuture future1 = service.getHttpData("http://www.jd.com"); 
      CompletableFuture future2 = service.getHttpData("http://www.jd.com"); 
      CompletableFuture future3 =service.getHttpData("http://www.jd.com"); 
      List<CompletableFuture> futureList = Lists.newArrayList(future1,future2, future3); 
      CompletableFuture<Void> allDoneFuture =CompletableFuture.allOf(futureList.toArray(newCompletableFuture[futureList.size()])); 
      CompletableFuture<String> future4 =allDoneFuture.thenApply(v -> { 
           List<Object> result =futureList.stream().map(CompletableFuture::join) 
                  .collect(Collectors.toList()); 
           //注意順序 
           String result1 = (String)result.get(0); 
           String result2 = (String)result.get(1); 
           String result3 = (String)result.get(2); 
           //處理業(yè)務(wù).... 
           return result1 + result2 + result3; 
       }).exceptionally(e -> { 
           //e.printStackTrace(); 
           return ""; 
       }); 
      //返回 
   } 

場(chǎng)景2、兩個(gè)服務(wù)并發(fā)異步調(diào)用，返回CompletableFuture，不阻塞主線程;

方法test2：

public void test2() throws Exception { 
      HelloClientDemoTest service = new HelloClientDemoTest(); 
      /** 
       * 場(chǎng)景2 兩個(gè)接口并發(fā)異步調(diào)用，返回CompletableFuture,不阻塞主線程 
       * 并且兩個(gè)服務(wù)也是異步非阻塞調(diào)用 
       */ 
      CompletableFuture future1 = service.getHttpData("http://www.jd.com"); 
      CompletableFuture future2 =service.getHttpData("http://www.jd.com"); 
      CompletableFuture future3 =future1.thenCombine(future2, (f1, f2) -> { 
           //處理業(yè)務(wù).... 
           return f1 + "," + f2; 
       }).exceptionally(e -> { 
           return ""; 
       }); 
      //返回 
   } 

場(chǎng)景3、兩個(gè)服務(wù)，并發(fā)異步調(diào)用兩個(gè)服務(wù)，并且一個(gè)服務(wù)的結(jié)果返回后再次調(diào)用另一服務(wù)，然后將三個(gè)結(jié)果后并處理，返回CompletableFuture，整個(gè)處理過程中不阻塞任何線程;

方法test3：

publicvoid test3() throws Exception { 
       HelloClientDemoTest service = new HelloClientDemoTest(); 
       /** 
        * 場(chǎng)景3 兩請(qǐng)求依賴調(diào)用，然后與另一服務(wù)結(jié)果組合處理，返回CompletableFuture,不阻塞主線程 
        * 并且兩個(gè)服務(wù)也是異步非阻塞調(diào)用 
        */ 
        CompletableFuture future1 = service.getHttpData("http://www.jd.com"); 
        CompletableFuture future2 = service.getHttpData("http://www.jd.com"); 
        CompletableFuture<String> future3= future1.thenApply((param) -> { 
            CompletableFuture future4 =service.getHttpData("http://www.jd.com"); 
            return future4; 
        }); 
        CompletableFuture future5 =future2.thenCombine(future3, (f2, f3) -> { 
            //....處理業(yè)務(wù) 
            return f2 + "," + f3; 
        }).exceptionally(e -> { 
            return ""; 
        }); 
        //返回future5 
    } 

全異步Web系統(tǒng)設(shè)計(jì)

主要技術(shù)：servlet3，JDK8 CompletableFuture，支持異步Callback調(diào)用的RPC框架。

先看一下處理流程圖：

servlet3：Servlet 接收到請(qǐng)求之后，可能首先需要對(duì)請(qǐng)求攜帶的數(shù)據(jù)進(jìn)行一些預(yù)處理;接著，Servlet 線程將請(qǐng)求轉(zhuǎn)交給一個(gè)異步線程來執(zhí)行業(yè)務(wù)處理，線程本身返回至容器。針對(duì)業(yè)務(wù)處理較耗時(shí)的情況，這將大大減少服務(wù)器資源的占用，并且提高并發(fā)處理速度。servlet3可參考商品詳情頁(yè)系統(tǒng)的Servlet3異步化實(shí)踐，結(jié)合其中講解的servlet3整合：

public void submitFuture(finalHttpServletRequest req, final Callable<CompletableFuture> task) throwsException{ 
       final String uri = req.getRequestURI(); 
       final Map<String, String[]> params = req.getParameterMap(); 
       final AsyncContext asyncContext = req.startAsync(); 
       asyncContext.getRequest().setAttribute("uri", uri); 
       asyncContext.getRequest().setAttribute("params", params); 
       asyncContext.setTimeout(asyncTimeoutInSeconds * 1000); 
       if(asyncListener != null) { 
           asyncContext.addListener(asyncListener); 
       } 
       CompletableFuture future = task.call(); 
       future.thenAccept(result -> { 
           HttpServletResponse resp = (HttpServletResponse)asyncContext.getResponse(); 
           try { 
                if(result instanceof String) { 
                    byte[] bytes = new byte[0]; 
                    if (StringUtils.isBlank(result)){ 
                       resp.setContentType("text/html;charset=gbk"); 
                       resp.setContentLength(0); 
                    } else { 
                        bytes =result.getBytes("GBK"); 
                    } 
                   //resp.setBufferSize(bytes.length); 
                   resp.setContentType("text/html;charset=gbk"); 
                   if(StringUtils.isNotBlank(localIp)) { 
                       resp.setHeader("t.ser", localIp); 
                    } 
                   resp.setContentLength(bytes.length); 
                   resp.getOutputStream().write(bytes); 
                } else { 
                    write(resp,JSONUtils.toJSON(result)); 
                } 
           } catch (Throwable e) { 
               resp.setStatus(HttpServletResponse.SC_INTERNAL_SERVER_ERROR); //程序內(nèi)部錯(cuò)誤 
                try { 
                    LOG.error("get infoerror, uri : {},  params : {}", uri,JSONUtils.toJSON(params), e); 
                } catch (Exception ex) { 
                } 
           } finally { 
                asyncContext.complete(); 
           } 
       }).exceptionally(e -> { 
           asyncContext.complete(); 
           return null; 
       }); 
} 

另外還有Java中協(xié)程庫(kù)Quasar，可參考《Java的纖程庫(kù) - Quasar》，目前沒有在應(yīng)用中使用并在測(cè)試FiberHttpServlet的時(shí)候遇到很多坑，日后把Quasar自如運(yùn)用后形成日記，希望能結(jié)實(shí)更多的朋友一起研究，踩坑。

作者：孫偉，目前負(fù)責(zé)京東商品詳情頁(yè)統(tǒng)一服務(wù)系統(tǒng)，寫過java，寫過ngx_lua，還寫過storm等，喜歡學(xué)習(xí)研究新事物。

【本文來自51CTO專欄作者張開濤的微信公眾號(hào)(開濤的博客)，公眾號(hào)id: kaitao-1234567】

 戳這里，看該作者更多好文