組合式異步編程

開啓線程執行任務，不論是使用Runnable（無返回值不支持上報異常）仍是Callable（有返回值支持上報異常）接口，均可以輕鬆實現。那麼若是是開啓線程池並須要獲取結果歸集的狀況下，如何實現，以及優劣，老司機直接看總結便可。

任務執行完，結果歸集時，幾種方式：

1.Futrue

 

原理:

以下圖，Future接口封裝了取消，獲取線程結果，以及狀態判斷是否取消，是否完成這幾個方法，都頗有用。

 

demo:

使用線程池提交Callable接口任務，返回Future接口，添加進list,最後遍歷FutureList且內部使用while輪詢,併發獲取結果

1 package thread.future;
 2 
 3 import java.util.ArrayList;
 4 import java.util.Date;
 5 import java.util.List;
 6 import java.util.concurrent.Callable;
 7 import java.util.concurrent.ExecutorService;
 8 import java.util.concurrent.Executors;
 9 import java.util.concurrent.Future;
10 
11 /**
12  * 
13  * @ClassName: FutureDemo
14  * @Description: Future多線程併發任務結果歸集
15  * @author denny.zhang
16  * @date 2016年11月4日 下午1:50:32
17  *
18  */
19 public class FutureDemo{
20 
21     public static void main(String[] args)  {
22         Long start = System.currentTimeMillis();
23         //開啓多線程
24         ExecutorService exs = Executors.newFixedThreadPool(10);
25         try {
26             //結果集
27             List<Integer> list = new ArrayList<Integer>();
28             List<Future<Integer>> futureList = new ArrayList<Future<Integer>>();
29             //1.高速提交10個任務，每一個任務返回一個Future入list
30             for(int i=0;i<10;i++){
31                 futureList.add(exs.submit(new CallableTask(i+1)));
32             }
33             Long getResultStart = System.currentTimeMillis();
34             System.out.println("結果歸集開始時間="+new Date());
35             //2.結果歸集，遍歷futureList,高速輪詢（模擬實現了併發）獲取future狀態成功完成後獲取結果，退出當前循環
36             for (Future<Integer> future : futureList) {
37                 while (true) {//CPU高速輪詢：每一個future都併發輪循，判斷完成狀態而後獲取結果，這一行，是本實現方案的精髓所在。即有10個future在高速輪詢，完成一個future的獲取結果，就關閉一個輪詢
38                     if (future.isDone()&& !future.isCancelled()) {//獲取future成功完成狀態，若是想要限制每一個任務的超時時間，取消本行的狀態判斷+future.get(1000*1, TimeUnit.MILLISECONDS)+catch超時異常使用便可。
39                         Integer i = future.get();//獲取結果
40                         System.out.println("任務i="+i+"獲取完成!"+new Date());
41                         list.add(i);
42                         break;//當前future獲取結果完畢，跳出while
43                     } else {
44                         Thread.sleep(1);//每次輪詢休息1毫秒（CPU納秒級），避免CPU高速輪循耗空CPU---》新手別忘記這個
45                     }
46                 }
47             }
48             System.out.println("list="+list);
49             System.out.println("總耗時="+(System.currentTimeMillis()-start)+",取結果歸集耗時="+(System.currentTimeMillis()-getResultStart));
50         } catch (Exception e) {
51             e.printStackTrace();
52         } finally {
53             exs.shutdown();
54         }
55     }
56     static class CallableTask implements Callable<Integer>{
57         Integer i;
58         
59         public CallableTask(Integer i) {
60             super();
61             this.i=i;
62         }
63 
64         @Override
65         public Integer call() throws Exception {
66             if(i==1){
67                 Thread.sleep(3000);//任務1耗時3秒
68             }else if(i==5){
69                 Thread.sleep(5000);//任務5耗時5秒
70             }else{
71                 Thread.sleep(1000);//其它任務耗時1秒
72             }
73             System.out.println("task線程："+Thread.currentThread().getName()+"任務i="+i+",完成！");  
74             return i;
75         }
76     }
77 }

下面我會根據線程池的線程數量+配合每一個線程設定不一樣的耗時時間，來看一下我們的線程執行細節。（不關心的能夠直接跳過~）

1.開啓定長爲10的線程池，ExecutorService exs = Executors.newFixedThreadPool(10);+任務1耗時3秒，任務5耗時5秒，其餘1秒。控制檯打印以下：

結果歸集開始時間=Fri Jun 16 08:38:10 CST 2017
task線程：pool-1-thread-3任務i=3,完成！+Fri Jun 16 08:38:11 CST 2017
task線程：pool-1-thread-4任務i=4,完成！+Fri Jun 16 08:38:11 CST 2017
task線程：pool-1-thread-2任務i=2,完成！+Fri Jun 16 08:38:11 CST 2017
task線程：pool-1-thread-7任務i=7,完成！+Fri Jun 16 08:38:11 CST 2017
task線程：pool-1-thread-10任務i=10,完成！+Fri Jun 16 08:38:11 CST 2017
task線程：pool-1-thread-8任務i=8,完成！+Fri Jun 16 08:38:11 CST 2017
task線程：pool-1-thread-9任務i=9,完成！+Fri Jun 16 08:38:11 CST 2017
task線程：pool-1-thread-6任務i=6,完成！+Fri Jun 16 08:38:11 CST 2017
task線程：pool-1-thread-1任務i=1,完成！+Fri Jun 16 08:38:13 CST 2017---》任務1執行耗時3秒
任務i=1獲取完成!Fri Jun 16 08:38:13 CST 2017
任務i=2獲取完成!Fri Jun 16 08:38:13 CST 2017
任務i=3獲取完成!Fri Jun 16 08:38:13 CST 2017
任務i=4獲取完成!Fri Jun 16 08:38:13 CST 2017
task線程：pool-1-thread-5任務i=5,完成！+Fri Jun 16 08:38:15 CST 2017---》任務5執行耗時5秒
任務i=5獲取完成!Fri Jun 16 08:38:15 CST 2017
任務i=6獲取完成!Fri Jun 16 08:38:15 CST 2017
任務i=7獲取完成!Fri Jun 16 08:38:15 CST 2017
任務i=8獲取完成!Fri Jun 16 08:38:15 CST 2017
任務i=9獲取完成!Fri Jun 16 08:38:15 CST 2017
任務i=10獲取完成!Fri Jun 16 08:38:15 CST 2017
list=[1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
總耗時=5006,取結果歸集耗時=5002---》符合邏輯，10個任務，定長10線程池，其中一個任務耗時3秒，一個任務耗時5秒，因爲併發高速輪訓，耗時取最長5秒

2.開啓定長爲5的線程池，ExecutorService exs = Executors.newFixedThreadPool(5);+任務1耗時3秒，任務5耗時5秒，其餘1秒。控制檯打印以下：

結果歸集開始時間=Fri Jun 16 08:42:45 CST 2017
task線程：pool-1-thread-4任務i=4,完成！+Fri Jun 16 08:42:46 CST 2017---》線程2,3,4第1波執行任務234，耗時1秒【當前任務】，總耗時1秒【所有任務】---》注：這裏的總耗時指的是最終完成所有10個任務的耗時時間線。
task線程：pool-1-thread-3任務i=3,完成！+Fri Jun 16 08:42:46 CST 2017
task線程：pool-1-thread-2任務i=2,完成！+Fri Jun 16 08:42:46 CST 2017
task線程：pool-1-thread-3任務i=6,完成！+Fri Jun 16 08:42:47 CST 2017---》線程2,3,4第2波執行任務678，耗時1秒，總耗時2秒
task線程：pool-1-thread-2任務i=7,完成！+Fri Jun 16 08:42:47 CST 2017
task線程：pool-1-thread-4任務i=8,完成！+Fri Jun 16 08:42:47 CST 2017
task線程：pool-1-thread-1任務i=1,完成！+Fri Jun 16 08:42:48 CST 2017---》線程1，第1波執行任務1，耗時3秒，總耗時3秒
task線程：pool-1-thread-3任務i=9,完成！+Fri Jun 16 08:42:48 CST 2017---》線程3，第3波執行任務9，耗時1秒，線程3總耗時3秒
任務i=1獲取完成!Fri Jun 16 08:42:48 CST 2017
task線程：pool-1-thread-2任務i=10,完成！+Fri Jun 16 08:42:48 CST 2017
任務i=2獲取完成!Fri Jun 16 08:42:48 CST 2017
任務i=3獲取完成!Fri Jun 16 08:42:48 CST 2017
任務i=4獲取完成!Fri Jun 16 08:42:48 CST 2017
task線程：pool-1-thread-5任務i=5,完成！+Fri Jun 16 08:42:50 CST 2017---》任務5，第一波執行任務5耗時5秒，總耗時5秒
任務i=5獲取完成!Fri Jun 16 08:42:50 CST 2017
任務i=6獲取完成!Fri Jun 16 08:42:50 CST 2017
任務i=7獲取完成!Fri Jun 16 08:42:50 CST 2017
任務i=8獲取完成!Fri Jun 16 08:42:50 CST 2017
任務i=9獲取完成!Fri Jun 16 08:42:50 CST 2017
任務i=10獲取完成!Fri Jun 16 08:42:50 CST 2017
list=[1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
總耗時=5006,取結果歸集耗時=5002------》符合邏輯，10個任務，定長5的線程池，因爲線程1阻塞3秒，線程5阻塞5秒，因爲併發高速輪訓，總耗時取最長5秒

3.開啓定長爲5的線程池且把線程sleep時間所有設定爲1秒，即ExecutorService exs = Executors.newFixedThreadPool(5);Thread.sleep(1000);控制檯打印以下：

1 結果歸集開始時間=Fri Jun 16 08:44:51 CST 2017
 2 task線程：pool-1-thread-4任務i=4,完成！+Fri Jun 16 08:44:52 CST 2017
 3 task線程：pool-1-thread-3任務i=3,完成！+Fri Jun 16 08:44:52 CST 2017
 4 task線程：pool-1-thread-5任務i=5,完成！+Fri Jun 16 08:44:52 CST 2017
 5 task線程：pool-1-thread-2任務i=2,完成！+Fri Jun 16 08:44:52 CST 2017
 6 task線程：pool-1-thread-1任務i=1,完成！+Fri Jun 16 08:44:52 CST 2017
 7 任務i=1獲取完成!Fri Jun 16 08:44:52 CST 2017
 8 任務i=2獲取完成!Fri Jun 16 08:44:52 CST 2017
 9 任務i=3獲取完成!Fri Jun 16 08:44:52 CST 2017
10 任務i=4獲取完成!Fri Jun 16 08:44:52 CST 2017
11 任務i=5獲取完成!Fri Jun 16 08:44:52 CST 2017
12 task線程：pool-1-thread-4任務i=6,完成！+Fri Jun 16 08:44:53 CST 2017
13 task線程：pool-1-thread-3任務i=7,完成！+Fri Jun 16 08:44:53 CST 2017
14 task線程：pool-1-thread-1任務i=10,完成！+Fri Jun 16 08:44:53 CST 2017
15 task線程：pool-1-thread-2任務i=9,完成！+Fri Jun 16 08:44:53 CST 2017
16 任務i=6獲取完成!Fri Jun 16 08:44:53 CST 2017
17 task線程：pool-1-thread-5任務i=8,完成！+Fri Jun 16 08:44:53 CST 2017
18 任務i=7獲取完成!Fri Jun 16 08:44:53 CST 2017
19 任務i=8獲取完成!Fri Jun 16 08:44:53 CST 2017
20 任務i=9獲取完成!Fri Jun 16 08:44:53 CST 2017
21 任務i=10獲取完成!Fri Jun 16 08:44:53 CST 2017
22 list=[1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
23 總耗時=2006,取結果歸集耗時=2002------》符合邏輯，10個任務，定長5的線程池，執行2波，耗時2*1秒=2秒

4.若是開啓定長爲10的線程池且每一個任務耗時一秒，即ExecutorService exs = Executors.newFixedThreadPool(10);Thread.sleep(1000);控制檯打印以下：

結果歸集開始時間=Fri Jun 16 08:47:36 CST 2017
task線程：pool-1-thread-8任務i=8,完成！+Fri Jun 16 08:47:37 CST 2017
task線程：pool-1-thread-2任務i=2,完成！+Fri Jun 16 08:47:37 CST 2017
task線程：pool-1-thread-9任務i=9,完成！+Fri Jun 16 08:47:37 CST 2017
task線程：pool-1-thread-5任務i=5,完成！+Fri Jun 16 08:47:37 CST 2017
task線程：pool-1-thread-6任務i=6,完成！+Fri Jun 16 08:47:37 CST 2017
task線程：pool-1-thread-1任務i=1,完成！+Fri Jun 16 08:47:37 CST 2017
task線程：pool-1-thread-10任務i=10,完成！+Fri Jun 16 08:47:37 CST 2017
task線程：pool-1-thread-4任務i=4,完成！+Fri Jun 16 08:47:37 CST 2017
task線程：pool-1-thread-7任務i=7,完成！+Fri Jun 16 08:47:37 CST 2017
task線程：pool-1-thread-3任務i=3,完成！+Fri Jun 16 08:47:37 CST 2017
任務i=1獲取完成!Fri Jun 16 08:47:37 CST 2017
任務i=2獲取完成!Fri Jun 16 08:47:37 CST 2017
任務i=3獲取完成!Fri Jun 16 08:47:37 CST 2017
任務i=4獲取完成!Fri Jun 16 08:47:37 CST 2017
任務i=5獲取完成!Fri Jun 16 08:47:37 CST 2017
任務i=6獲取完成!Fri Jun 16 08:47:37 CST 2017
任務i=7獲取完成!Fri Jun 16 08:47:37 CST 2017
任務i=8獲取完成!Fri Jun 16 08:47:37 CST 2017
任務i=9獲取完成!Fri Jun 16 08:47:37 CST 2017
任務i=10獲取完成!Fri Jun 16 08:47:37 CST 2017
list=[1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
總耗時=1009,取結果歸集耗時=1005------》符合邏輯，10個任務，定長10的線程池，執行1波，耗時1秒

 

建議：此種方法可實現基本目標，任務並行且按照提交順序獲取結果。使用很廣泛，老小皆宜，就是CPU有消耗，可使用！

2.FutureTask

 

原理:

是接口RunnableFuture的惟一實現類。類圖以下（網上截取來的。。。個人eclipse類圖插件還沒裝好）：

 

如上圖，可見RunnableFuture接口繼承自Future<V>+Runnable:

1.Runnable接口，可開啓單個線程執行。

2.Future<v>接口，可接受Callable接口的返回值，futureTask.get()阻塞獲取結果。

FutureTask的構造方法有兩種，其實最終都是賦值callable。以下圖：

demo:

demo1:兩個步驟：1.開啓單個線程執行任務，2.阻塞等待執行結果，分離這兩步驟，可在這兩步中間穿插別的相關業務邏輯。

1 /**
 2  * 
 3  * @ClassName:FutureTaskDemo
 4  * @Description:FutureTask彌補了Future必須用線程池提交返回Future的缺陷，實現功能以下：
 5  * 1.Runnable接口，可開啓線程執行。
 6  * 2.Future<v>接口，可接受Callable接口的返回值，futureTask.get()阻塞獲取結果。
 7  * 這兩個步驟：一個開啓線程執行任務，一個阻塞等待執行結果，分離這兩步驟，可在這兩步中間穿插別的相關業務邏輯。
 8  * @author diandian.zhang
 9  * @date 2017年6月16日上午10:36:05
10  */
11 public class FutureTaskContorlDemo {
12     
13     public static void main(String[] args)  {
14         try {
15             System.out.println("=====例如一個統計公司總部和分部的總利潤是否達標100萬==========");
16             //利潤
17             Integer count = 0;
18             //1.定義一個futureTask，假設去遠程http獲取各個分公司業績.
19             FutureTask<Integer> futureTask = new FutureTask<Integer>(new CallableTask());
20             Thread futureTaskThread =  new Thread(futureTask);
21             futureTaskThread.start();
22             System.out.println("futureTaskThread start！"+new Date());
23             
24             //2.主線程先作點別的事
25             System.out.println("主線程查詢總部公司利潤開始時間："+new Date());
26             Thread.sleep(5000);
27             count+=10;//北京集團總部利潤。
28             System.out.println("主線程查詢總部公司利潤結果時間："+new Date());
29             
30             //總部已達標100萬利潤，就再也不繼續執行獲取分公司業績任務了
31             if(count>=100){
32                 System.out.println("總部公司利潤達標，取消futureTask！"+new Date());
33                 futureTask.cancel(true);//不須要再去獲取結果，那麼直接取消便可
34             }else{
35                 System.out.println("總部公司利潤未達標，進入阻塞查詢分公司利潤！"+new Date());
36                 //3總部未達標.阻塞獲取，各個分公司結果
37                 Integer i = futureTask.get();//真正執行CallableTask
38                 System.out.println("i="+i+"獲取到結果!"+new Date()+new Date());
39             }
40         } catch (Exception e) {
41             e.printStackTrace();
42         }
43     }
44 
45     /**
46      * 
47      * @ClassName:CallableTask
48      * @Description:一個十分耗時的任務
49      * @author diandian.zhang
50      * @date 2017年6月16日上午10:39:04
51      */
52     static class CallableTask implements Callable<Integer>{
53         @Override
54         public Integer call() throws Exception {
55             System.out.println("CallableTask-call，查詢分公司利潤，執行開始！"+new Date());
56             Thread.sleep(10000);
57             System.out.println("CallableTask-call，查詢分公司利潤，執行完畢！"+new Date());
58             return 10;
59         }
60     }

執行結果以下：

=====例如一個統計公司總部和分部的總利潤是否達標100萬==========
futureTaskThread start！Fri Jun 16 11:14:54 CST 2017----》futureTaskThread 已開始運行
CallableTask-call，查詢分公司利潤，執行開始！Fri Jun 16 11:14:54 CST 2017
主線程查詢總部公司利潤開始時間：Fri Jun 16 11:14:54 CST 2017------》主線程耗時5秒
主線程查詢總部公司利潤結果時間：Fri Jun 16 11:14:59 CST 2017
總部公司利潤未達標，進入阻塞查詢分公司利潤！Fri Jun 16 11:14:59 CST 2017
CallableTask-call，查詢分公司利潤，執行完畢！Fri Jun 16 11:15:04 CST 2017----》futureTaskThread 執行完畢，耗時10秒
i=10獲取到結果!Fri Jun 16 11:15:04 CST 2017Fri Jun 16 11:15:04 CST 2017

如上圖，分離以後，futureTaskThread耗時10秒期間，主線程還穿插的執行了耗時5秒的操做，大大減少總耗時。且可根據業務邏輯實時判斷是否須要繼續執行futureTask。

demo2：固然FutureTask同樣能夠併發執行任務並獲取結果，以下：

1 package thread.future;
 2 
 3 import java.util.ArrayList;
 4 import java.util.Date;
 5 import java.util.List;
 6 import java.util.concurrent.Callable;
 7 import java.util.concurrent.ExecutorService;
 8 import java.util.concurrent.Executors;
 9 import java.util.concurrent.FutureTask;
10 
11 /**
12  * 
13  * @ClassName:FutureTaskDemo
14  * @Description:FutureTask實現多線程併發執行任務並取結果歸集
15  * @author diandian.zhang
16  * @date 2017年6月16日上午10:36:05
17  */
18 public class FutureTaskDemo {
19     
20     public static void main(String[] args)  {
21         Long start = System.currentTimeMillis();
22         //開啓多線程
23         ExecutorService exs = Executors.newFixedThreadPool(5);
24         try {
25             //結果集
26             List<Integer> list = new ArrayList<Integer>();
27             List<FutureTask<Integer>> futureList = new ArrayList<FutureTask<Integer>>();
28             //啓動線程池，10個任務固定線程數爲5
29             for(int i=0;i<10;i++){
30                  FutureTask<Integer> futureTask = new FutureTask<Integer>(new CallableTask(i+1));
31                 //提交任務，添加返回
32                 exs.submit(futureTask);//Runnable特性
33                 futureList.add(futureTask);//Future特性
34             }
35             Long getResultStart = System.currentTimeMillis();
36             System.out.println("結果歸集開始時間="+new Date());
37             //結果歸集
38             for (FutureTask<Integer> future : futureList) {
39                 while (true) {
40                     if (future.isDone()&& !future.isCancelled()) {
41                         Integer i = future.get();//Future特性
42                         System.out.println("i="+i+"獲取到結果!"+new Date());
43                         list.add(i);
44                         break;
45                     }else {
46                         Thread.sleep(1);//避免CPU高速輪循，能夠休息一下。
47                     }
48                 }
49             }
50             System.out.println("list="+list);
51             System.out.println("總耗時="+(System.currentTimeMillis()-start)+",取結果歸集耗時="+(System.currentTimeMillis()-getResultStart));
52         } catch (Exception e) {
53             e.printStackTrace();
54         } finally {
55             exs.shutdown();
56         }
57         
58     }
59 
60     static class CallableTask implements Callable<Integer>{
61         Integer i;
62         
63         public CallableTask(Integer i) {
64             super();
65             this.i=i;
66         }
67 
68         @Override
69         public Integer call() throws Exception {
70             if(i==1){
71                 Thread.sleep(3000);//任務1耗時3秒
72             }else if(i==5){
73                 Thread.sleep(5000);//任務5耗時5秒
74             }else{
75                 Thread.sleep(1000);//其它任務耗時1秒
76             }
77             System.out.println("線程：["+Thread.currentThread().getName()+"]任務i="+i+",完成！"+new Date());  
78             return i;
79 
80         }
81         
82     }
83 
84 }

運行結果就不打印了，和Future同樣的。由於結果歸集用了Future<v>特性。

 

建議：demo1在特定場合例若有十分耗時的業務但有依賴於其餘業務不必定非要執行的，能夠嘗試使用。demo2多線程併發執行並結果歸集，這裏多套一層FutureTask比較雞肋（直接返回Future簡單明瞭）不建議使用。

3.CompletionService：

原理:內部經過阻塞隊列+FutureTask，實現了任務先完成可優先獲取到，即結果按照完成前後順序排序。

demo:

1 package thread.future;
 2 
 3 import java.util.ArrayList;
 4 import java.util.Date;
 5 import java.util.List;
 6 import java.util.concurrent.Callable;
 7 import java.util.concurrent.CompletionService;
 8 import java.util.concurrent.ExecutorCompletionService;
 9 import java.util.concurrent.ExecutorService;
10 import java.util.concurrent.Executors;
11 import java.util.concurrent.Future;
12 
13 /**
14  * 
15  * @ClassName: CompletionServiceDemo
16  * @Description: CompletionService多線程併發任務結果歸集
17  * @author denny.zhang
18  * @date 2016年11月4日 下午1:50:32
19  *
20  */
21 public class CompletionServiceDemo{
22 
23     public static void main(String[] args)  {
24         Long start = System.currentTimeMillis();
25         //開啓3個線程
26         ExecutorService exs = Executors.newFixedThreadPool(5);
27         try {
28             int taskCount = 10;
29             //結果集
30             List<Integer> list = new ArrayList<Integer>();
31             //1.定義CompletionService
32             CompletionService<Integer> completionService = new ExecutorCompletionService<Integer>(exs);  
33             List<Future<Integer>> futureList = new ArrayList<Future<Integer>>();
34             //2.添加任務
35             for(int i=0;i<taskCount;i++){
36                 futureList.add(completionService.submit(new Task(i+1)));
37             }
38             //==================結果歸集===================
39             //方法1：future是提交時返回的，遍歷queue則按照任務提交順序，獲取結果
40 //            for (Future<Integer> future : futureList) {
41 //                System.out.println("====================");
42 //                Integer result = future.get();//線程在這裏阻塞等待該任務執行完畢,按照
43 //                System.out.println("任務result="+result+"獲取到結果!"+new Date());
44 //                list.add(result);
45 //            }
46 
47 //            //方法2.使用內部阻塞隊列的take()
48             for(int i=0;i<taskCount;i++){
49                 Integer result = completionService.take().get();//採用completionService.take()，內部維護阻塞隊列，任務先完成的先獲取到
50                 System.out.println("任務i=="+result+"完成!"+new Date());
51                 list.add(result);
52             }
53             System.out.println("list="+list);
54             System.out.println("總耗時="+(System.currentTimeMillis()-start));
55         } catch (Exception e) {
56             e.printStackTrace();
57         } finally {
58             exs.shutdown();//關閉線程池
59         }
60         
61     }
62 
63     static class Task implements Callable<Integer>{
64         Integer i;
65         
66         public Task(Integer i) {
67             super();
68             this.i=i;
69         }
70 
71         @Override
72         public Integer call() throws Exception {
73             if(i==5){
74                 Thread.sleep(5000);
75             }else{
76                 Thread.sleep(1000);
77             }
78             System.out.println("線程："+Thread.currentThread().getName()+"任務i="+i+",執行完成！");  
79             return i;
80         }
81         
82     }
83 }

打印結果以下：

線程：pool-1-thread-3任務i=3,執行完成！
線程：pool-1-thread-1任務i=1,執行完成！
線程：pool-1-thread-4任務i=4,執行完成！
線程：pool-1-thread-2任務i=2,執行完成！
任務i==3完成!Fri Jun 16 11:39:17 CST 2017
任務i==1完成!Fri Jun 16 11:39:17 CST 2017
任務i==4完成!Fri Jun 16 11:39:17 CST 2017
任務i==2完成!Fri Jun 16 11:39:17 CST 2017
線程：pool-1-thread-4任務i=8,執行完成！
線程：pool-1-thread-3任務i=7,執行完成！
線程：pool-1-thread-1任務i=6,執行完成！
線程：pool-1-thread-2任務i=9,執行完成！
任務i==8完成!Fri Jun 16 11:39:18 CST 2017
任務i==7完成!Fri Jun 16 11:39:18 CST 2017
任務i==6完成!Fri Jun 16 11:39:18 CST 2017
任務i==9完成!Fri Jun 16 11:39:18 CST 2017
線程：pool-1-thread-3任務i=10,執行完成！
任務i==10完成!Fri Jun 16 11:39:19 CST 2017
線程：pool-1-thread-5任務i=5,執行完成！
任務i==5完成!Fri Jun 16 11:39:21 CST 2017
list=[3, 1, 4, 2, 8, 7, 6, 9, 10, 5]---》這裏證明了確實按照執行完成順序排序
總耗時=5004---》符合邏輯，10個任務，定長5線程池執行，取最長時間。

建議：使用率也挺高，並且能按照完成前後排序，建議若是有排序需求的優先使用。只是多線程併發執行任務結果歸集，也可使用。

4.CompletableFuture

原理：

4.1.從註釋看：

JDK1.8才新加入的一個實現類，實現了Future<T>, CompletionStage<T>2個接口，JDK註釋以下圖：

譯文（沒興趣的能夠跳過）：

當一個Future可能須要顯示地完成時，使用CompletionStage接口去支持完成時觸發的函數和操做。當2個以上線程同時嘗試完成、異常完成、取消一個CompletableFuture時，只有一個能成功。

CompletableFuture實現了CompletionStage接口的以下策略：

1.爲了完成當前的CompletableFuture接口或者其餘完成方法的回調函數的線程，提供了非異步的完成操做。

2.沒有顯式入參Executor的全部async方法都使用ForkJoinPool.commonPool()爲了簡化監視、調試和跟蹤，全部生成的異步任務都是標記接口AsynchronousCompletionTask的實例。

3.全部的CompletionStage方法都是獨立於其餘共有方法實現的，所以一個方法的行爲不會受到子類中其餘方法的覆蓋。

CompletableFuture實現了Futurre接口的以下策略：

1.CompletableFuture沒法直接控制完成，因此cancel操做被視爲是另外一種異常完成形式。方法isCompletedExceptionally能夠用來肯定一個CompletableFuture是否以任何異常的方式完成。

2.以一個CompletionException爲例，方法get()和get(long,TimeUnit)拋出一個ExecutionException，對應CompletionException。爲了在大多數上下文中簡化用法，這個類還定義了方法join()和getNow，而不是直接在這些狀況中直接拋出CompletionException。

 

4.2.CompletionStage接口實現流式編程：

JDK8新增接口，此接口包含38個方法...是的，你沒看錯，就是38個方法。這些方法主要是爲了支持函數式編程中流式處理。

 

4.3.CompletableFuture中4個異步執行任務靜態方法：

如上圖，其中supplyAsync用於有返回值的任務，runAsync則用於沒有返回值的任務。Executor參數能夠手動指定線程池，不然默認ForkJoinPool.commonPool()系統級公共線程池，注意：這些線程都是Daemon線程，主線程結束Daemon線程不結束，只有JVM關閉時，生命週期終止。

4.4.組合CompletableFuture：

thenCombine()： 先完成當前CompletionStage和other 2個CompletionStage任務，而後把結果傳參給BiFunction進行結果合併操做。

 

thenCompose()：第一個CompletableFuture執行完畢後，傳遞給下一個CompletionStage做爲入參進行操做。

demo:

JDK CompletableFuture  自帶多任務組合方法allOf和anyOf

allOf是等待全部任務完成，構造後CompletableFuture完成

anyOf是隻要有一個任務完成，構造後CompletableFuture就完成

方式一：循環建立CompletableFuture list,調用sequence()組裝返回一個有返回值的CompletableFuture，返回結果get()獲取

方式二：全流式處理轉換成CompletableFuture[ ] + allOf 組裝成一個無返回值CompletableFuture，join等待執行完畢。返回結果whenComplete獲取。---》推薦

1 package thread.future;
  2 
  3 import java.util.ArrayList;
  4 import java.util.Date;
  5 import java.util.List;
  6 import java.util.concurrent.CompletableFuture;
  7 import java.util.concurrent.ExecutorService;
  8 import java.util.concurrent.Executors;
  9 import java.util.stream.Collectors;
 10 import java.util.stream.Stream;
 11 
 12 import com.google.common.collect.Lists;
 13 
 14 /**
 15  * 
 16  * @ClassName:CompletableFutureDemo
 17  * @Description:多線程併發任務,取結果歸集
 18  * @author diandian.zhang
 19  * @date 2017年6月14日下午12:44:01
 20  */
 21 public class CompletableFutureDemo {
 22     public static void main(String[] args) {
 23         Long start = System.currentTimeMillis();
 24         //結果集
 25         List<String> list = new ArrayList<String>();
 26         List<String> list2 = new ArrayList<String>();
 27         //定長10線程池
 28         ExecutorService exs = Executors.newFixedThreadPool(10);
 29         //List<CompletableFuture<String>> futureList = new ArrayList<>();
 30         List<Integer> taskList = Lists.newArrayList(2,1,3,4,5,6,7,8,9,10);
 31         try {
 32             //方式一：循環建立CompletableFuture list,調用sequence()組裝返回一個有返回值的CompletableFuture，返回結果get()獲取
 33 //            for(int i=0;i<taskList.size();i++){
 34 //                final int j=i+1;
 35 //                CompletableFuture<String> future = CompletableFuture.supplyAsync(()->calc(j), exs)//異步執行
 36 //                        .thenApply(e->Integer.toString(e))//Integer轉換字符串    thenAccept只接受不返回不影響結果
 37 ////                        .whenComplete((v, e) -> {//如需獲取任務完成先手順序，此處代碼便可
 38 ////                            System.out.println("任務"+v+"完成!result="+v+"，異常 e="+e+","+new Date());
 39 ////                            list2.add(v);
 40 ////                        })
 41 //                        ;
 42 //                futureList.add(future);
 43 //            }
 44 //            //流式獲取結果
 45 //            list = sequence(futureList).get();//[1, 2, 3, 4, 5, 6, 7, 8, 9, 10]此處不理解爲何是這樣的順序？誰知道求告知
 46             
 47             //方式二：全流式處理轉換成CompletableFuture[]+組裝成一個無返回值CompletableFuture，join等待執行完畢。返回結果whenComplete獲取
 48             @SuppressWarnings("rawtypes")
 49             CompletableFuture[] cfs = taskList.stream().map(object-> CompletableFuture.supplyAsync(()->calc(object), exs)
 50                     .thenApply(h->Integer.toString(h))
 51                     .whenComplete((v, e) -> {//如需獲取任務完成先手順序，此處代碼便可
 52                         System.out.println("任務"+v+"完成!result="+v+"，異常 e="+e+","+new Date());
 53                         list2.add(v);
 54                     }))
 55                     .toArray(CompletableFuture[]::new);
 56             CompletableFuture.allOf(cfs).join();//封裝後無返回值，必須本身whenComplete()獲取
 57             System.out.println("list2="+list2+"list="+list+",耗時="+(System.currentTimeMillis()-start));
 58         } catch (Exception e) {
 59             e.printStackTrace();
 60         }finally {
 61             exs.shutdown();
 62         }
 63     }
 64     
 65     public static Integer calc(Integer i){
 66         try {
 67             if(i==1){
 68                 Thread.sleep(3000);//任務1耗時3秒
 69             }else if(i==5){
 70                 Thread.sleep(5000);//任務5耗時5秒
 71             }else{
 72                 Thread.sleep(1000);//其它任務耗時1秒
 73             }
 74             System.out.println("task線程："+Thread.currentThread().getName()+"任務i="+i+",完成！+"+new Date());  
 75         } catch (InterruptedException e) {
 76             e.printStackTrace();
 77         }
 78         return i;
 79     }
 80     
 81     /**
 82      * 
 83      * @Description 組合多個CompletableFuture爲一個CompletableFuture,全部子任務所有完成，組合後的任務纔會完成。帶返回值，可直接get.
 84      * @param futures List
 85      * @return
 86      * @author diandian.zhang
 87      * @date 2017年6月19日下午3:01:09
 88      * @since JDK1.8
 89      */
 90     public static <T> CompletableFuture<List<T>> sequence(List<CompletableFuture<T>> futures) {
 91         //1.構造一個空CompletableFuture，子任務數爲入參任務list size
 92         CompletableFuture<Void> allDoneFuture = CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()]));
 93         //2.流式（每一個子任務join操做後轉換爲list）往空CompletableFuture中添加結果
 94         return allDoneFuture.thenApply(v -> futures.stream().map(CompletableFuture::join).collect(Collectors.<T>toList()));
 95     }
 96    
 97     /**
 98      * 
 99      * @Description Stream流式類型futures轉換成一個CompletableFuture,全部子任務所有完成，組合後的任務纔會完成。帶返回值，可直接get.
100      * @param futures Stream
101      * @return
102      * @author diandian.zhang
103      * @date 2017年6月19日下午6:23:40
104      * @since JDK1.8
105      */
106     public static <T> CompletableFuture<List<T>> sequence(Stream<CompletableFuture<T>> futures) {
107         List<CompletableFuture<T>> futureList = futures.filter(f -> f != null).collect(Collectors.toList());
108         return sequence(futureList);
109     }
110 }

方式二返回結果：

task線程：pool-1-thread-1任務i=2,完成！+Mon Jun 19 18:26:17 CST 2017
task線程：pool-1-thread-9任務i=9,完成！+Mon Jun 19 18:26:17 CST 2017
task線程：pool-1-thread-6任務i=6,完成！+Mon Jun 19 18:26:17 CST 2017
task線程：pool-1-thread-8任務i=8,完成！+Mon Jun 19 18:26:17 CST 2017
任務6完成!result=6，異常 e=null,Mon Jun 19 18:26:17 CST 2017
task線程：pool-1-thread-4任務i=4,完成！+Mon Jun 19 18:26:17 CST 2017
task線程：pool-1-thread-7任務i=7,完成！+Mon Jun 19 18:26:17 CST 2017
任務4完成!result=4，異常 e=null,Mon Jun 19 18:26:17 CST 2017
task線程：pool-1-thread-3任務i=3,完成！+Mon Jun 19 18:26:17 CST 2017
任務3完成!result=3，異常 e=null,Mon Jun 19 18:26:17 CST 2017
task線程：pool-1-thread-10任務i=10,完成！+Mon Jun 19 18:26:17 CST 2017
任務10完成!result=10，異常 e=null,Mon Jun 19 18:26:17 CST 2017
任務7完成!result=7，異常 e=null,Mon Jun 19 18:26:17 CST 2017
任務8完成!result=8，異常 e=null,Mon Jun 19 18:26:17 CST 2017
任務2完成!result=2，異常 e=null,Mon Jun 19 18:26:17 CST 2017
任務9完成!result=9，異常 e=null,Mon Jun 19 18:26:17 CST 2017
task線程：pool-1-thread-2任務i=1,完成！+Mon Jun 19 18:26:19 CST 2017---》任務1耗時3秒
任務1完成!result=1，異常 e=null,Mon Jun 19 18:26:19 CST 2017
task線程：pool-1-thread-5任務i=5,完成！+Mon Jun 19 18:26:21 CST 2017---》任務5耗時5秒
任務5完成!result=5，異常 e=null,Mon Jun 19 18:26:21 CST 2017
list2=[6, 4, 3, 10, 7, 8, 2, 9, 1, 5]list=[],耗時=5076---》符合邏輯，10個任務，10個線程併發執行，其中任務1耗時3秒，任務5耗時5秒，耗時取最大值。

建議：CompletableFuture知足併發執行，順序完成先手順序獲取的目標。並且支持每一個任務的異常返回，配合流式編程，用起來速度飛起。JDK源生支持，API豐富，推薦使用。

5.總結：

本文從原理、demo、建議三個方向分析了經常使用多線程併發，取結果歸集的幾種實現方案，但願對你們有所啓發，整理表格以下：

	Futrue	FutureTask	CompletionService	CompletableFuture
原理	Futrue接口	接口RunnableFuture的惟一實現類，RunnableFuture接口繼承自Future<V>+Runnable:	內部經過阻塞隊列+FutureTask接口	JDK8實現了Future<T>, CompletionStage<T>2個接口
多任務併發執行	支持	支持	支持	支持
獲取任務結果的順序	按照提交順序獲取結果	未知	支持任務完成的前後順序	支持任務完成的前後順序
異常捕捉	本身捕捉	本身捕捉	本身捕捉	源生API支持，返回每一個任務的異常
建議	CPU高速輪詢，耗資源，可使用，但不推薦	功能不對口，併發任務這一塊多套一層，不推薦使用。	推薦使用，沒有JDK8CompletableFuture以前最好的方案，沒有質疑	API極端豐富，配合流式編程，速度飛起，推薦使用！

=====參考=================

1.《JAVA高併發程序設計》

2.使用Java 8的CompletableFuture實現函數式的回調

3.Java CompletableFuture 詳解