多線程系列六：線程池

時間 2019-11-12

標籤多線程系列線程欄目 Java 简体版

原文原文鏈接

一. 線程池簡介

1. 線程池的概念：

線程池就是首先建立一些線程，它們的集合稱爲線程池。java

2. 使用線程池的好處

a) 下降資源的消耗。使用線程池不用頻繁的建立線程和銷燬線程程序員

b) 提升響應速度，任務：T1建立線程時間，T2任務執行時間，T3線程銷燬時間，線程池空閒的時候能夠去執行T1和T2，從而提升響應編程

c) 提升線程的可管理性。小程序

使用線程池能夠很好地提升性能，線程池在系統啓動時即建立大量空閒的線程，程序將一個任務傳給線程池，線程池就會啓動一條線程來執行這個任務，執行結束之後，該線程並不會死亡，而是再次返回線程池中成爲空閒狀態，等待執行下一個任務。服務器

3. 線程池的工做機制

2.1 在線程池的編程模式下，任務是提交給整個線程池，而不是直接提交給某個線程，線程池在拿到任務後，就在內部尋找是否有空閒的線程，若是有，則將任務交給某個空閒的線程。多線程

2.1 一個線程同時只能執行一個任務，但能夠同時向一個線程池提交多個任務。併發

4. 使用線程池的緣由：

多線程運行時間，系統不斷的啓動和關閉新線程，成本很是高，會過渡消耗系統資源，以及過渡切換線程的危險，從而可能致使系統資源的崩潰。這時，線程池就是最好的選擇了。框架

5. 線程池的主要處理流程

說明：dom

a）線程池判斷核心線程池裏的線程是否都在執行任務。若是不是，則建立一個新的工做線程來執行任務。若是核心線程池裏的線程都在執行任務，則進入下個流程b。異步

b）線程池判斷工做隊列是否已經滿。若是工做隊列沒有滿，則將新提交的任務存儲在這個工做隊列裏。若是工做隊列滿了，則進入下個流程c。

c）線程池判斷線程池的線程是否都處於工做狀態。若是沒有，則建立一個新的工做線程來執行任務。若是已經滿了，則交給飽和策略來處理這個任務。

6. ThreadPoolExecutor執行execute()方法的示意

執行execute()方法是對第5點中的線程池的主要處理流程的更深層次的說明

a）若是當前運行的線程少於corePoolSize，則建立新線程來執行任務（注意，執行這一步驟須要獲取全局鎖）。

b）若是運行的線程等於或多於corePoolSize，則將任務加入BlockingQueue。

c）若是沒法將任務加入BlockingQueue（隊列已滿），則建立新的線程來處理任務（注意，執行這一步驟須要獲取全局鎖）。

d）若是建立新線程將使當前運行的線程超出maximumPoolSize，任務將被拒絕，並調用RejectedExecutionHandler.rejectedExecution()方法。

7.線程池的建立各個參數含義

public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler)

corePoolSize

線程池中的核心線程數，當提交一個任務時，線程池建立一個新線程執行任務，直到當前線程數等於corePoolSize；

若是當前線程數爲corePoolSize，繼續提交的任務被保存到阻塞隊列中，等待被執行；

若是執行了線程池的prestartAllCoreThreads()方法，線程池會提早建立並啓動全部核心線程。

maximumPoolSize

線程池中容許的最大線程數。若是當前阻塞隊列滿了，且繼續提交任務，則建立新的線程執行任務，前提是當前線程數小於maximumPoolSize

keepAliveTime

線程空閒時的存活時間，即當線程沒有任務執行時，繼續存活的時間。默認狀況下，該參數只在線程數大於corePoolSize時纔有用

TimeUnit

keepAliveTime的時間單位

workQueue

workQueue必須是BlockingQueue阻塞隊列。當線程池中的線程數超過它的corePoolSize的時候，線程會進入阻塞隊列進行阻塞等待。經過workQueue，線程池實現了阻塞功能

threadFactory

建立線程的工廠，經過自定義的線程工廠能夠給每一個新建的線程設置一個具備識別度的線程名

Executors靜態工廠裏默認的threadFactory，線程的命名規則是「pool-數字-thread-數字」

8.RejectedExecutionHandler（飽和策略）

線程池的飽和策略，當阻塞隊列滿了，且沒有空閒的工做線程，若是繼續提交任務，必須採起一種策略處理該任務，線程池提供了4種策略：

（1）AbortPolicy：直接拋出異常，默認策略；

（2）CallerRunsPolicy：用調用者所在的線程來執行任務；

（3）DiscardOldestPolicy：丟棄阻塞隊列中靠最前的任務，並執行當前任務；

（4）DiscardPolicy：直接丟棄任務；

固然也能夠根據應用場景實現RejectedExecutionHandler接口，自定義飽和策略，如記錄日誌或持久化存儲不能處理的任務。

9.關閉線程池

shutDown():interrupt方法來終止線程

shutDownNow() 嘗試中止全部正在執行的線程

10. 合理地配置線程池

線程數配置：

任務：計算密集型，IO密集型，混合型

計算密集型適合配置的線程數=計算機的cpu數或計算機的cpu數+1（應付頁缺失）

IO密集型適合配置的線程數=計算機的cpu數*2

混合型適合配置的線程數，拆分紅計算密集型，IO密集型

Runtime.getRuntime().availableProcessors();當前機器中的cpu核心個數

隊列的選擇：

儘可能有界隊列，不要使用無界隊列

2、使用jdk中線程池的案例

按 Ctrl+C 複製代碼

import java.util.Random;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
* 使用jdk中線程池的案例
*/
public class UseThreadPool {

static class MyTask implements Runnable {

private String name;

public MyTask(String name) {
this.name = name;
}

public String getName() {
return name;
}

@Override
public void run() {// 執行任務
try {
Random r = new Random();
Thread.sleep(r.nextInt(1000)+2000);
} catch (InterruptedException e) {
System.out.println(Thread.currentThread().getId()+" sleep InterruptedException:"
+Thread.currentThread().isInterrupted());
}
System.out.println("任務 " + name + " 完成");
}
}

public static void main(String[] args) {
//建立線程池
ThreadPoolExecutor threadPoolExecutor =
new ThreadPoolExecutor(2,4,60,
TimeUnit.SECONDS,new ArrayBlockingQueue<Runnable>(10));
//往線程池裏面提交6個線程去執行
for(int i =0;i<=5;i++){
MyTask task = new MyTask("Task_"+i);
System.out.println("A new task will add:"+task.getName());
threadPoolExecutor.execute(task);

}
//關閉線程池
threadPoolExecutor.shutdown();
}

}

按 Ctrl+C 複製代碼

3、實現本身的一個線程池

手寫的線程池MyThreadPool

按 Ctrl+C 複製代碼

import java.util.LinkedList;
import java.util.List;

/**
* 實現本身的一個線程池
*/
public class MyThreadPool {

//默認的線程個數
private int work_num = 5;

//線程的容器
private WorkThread[] workThreads;

//任務隊列
private List<Runnable> taskQueue = new LinkedList<>();

public MyThreadPool(int work_num) {
this.work_num = work_num;
workThreads = new WorkThread[work_num];
for(int i=0;i<work_num;i++){
workThreads[i] = new WorkThread();
workThreads[i].start();
}
}

//提交任務的接口
public void execute(Runnable task){
synchronized (taskQueue){
taskQueue.add(task);
taskQueue.notify();
}
}

//銷燬線程池
public void destroy(){
System.out.println("ready stop pool....");
for(int i=0;i<work_num;i++){
workThreads[i].stopWorker();
workThreads[i] = null;//加速垃圾回收
}
taskQueue.clear();
}

//工做線程
private class WorkThread extends Thread{

private volatile boolean on = true;

public void run(){
Runnable r = null;
try{
while(on&&!isInterrupted()){
synchronized (taskQueue){
//任務隊列中無任務，工做線程等待
while(on&&!isInterrupted()&&taskQueue.isEmpty()){
taskQueue.wait(1000);
}
//任務隊列中有任務，拿任務作事
if(on&&!isInterrupted()&&!taskQueue.isEmpty()){
r = taskQueue.remove(0);
}
}
if (r!=null){
System.out.println(getId()+" ready execute....");
r.run();
}
//加速垃圾回收
r = null;
}

}catch(InterruptedException e){
System.out.println(Thread.currentThread().getId()+" is Interrupted");
}
}

public void stopWorker(){
on = false;
interrupt();
}

}

按 Ctrl+C 複製代碼

測試手寫實現的線程池TestMyThreadPool

按 Ctrl+C 複製代碼

import java.util.Random;

/**
* 測試手寫實現的線程池
*/
public class TestMyThreadPool {
public static void main(String[] args) throws InterruptedException {
// 建立3個線程的線程池
MyThreadPool t = new MyThreadPool(3);
t.execute(new MyTask("testA"));
t.execute(new MyTask("testB"));
t.execute(new MyTask("testC"));
t.execute(new MyTask("testD"));
t.execute(new MyTask("testE"));
System.out.println(t);
Thread.sleep(3000);
t.destroy();// 全部線程都執行完成才destory
System.out.println(t);
}

// 任務類
static class MyTask implements Runnable {

private String name;
private Random r = new Random();

public MyTask(String name) {
this.name = name;
}

public String getName() {
return name;
}

@Override
public void run() {// 執行任務
try {
Thread.sleep(r.nextInt(1000)+2000);
} catch (InterruptedException e) {
System.out.println(Thread.currentThread().getId()+" sleep InterruptedException:"
+Thread.currentThread().isInterrupted());
}
System.out.println("任務 " + name + " 完成");
}
}
}

按 Ctrl+C 複製代碼

4、線程池框架Executor框架

1. Executor框架調度模型

在HotSpot VM的線程模型中，Java線程（java.lang.Thread）被一對一映射爲本地操做系統線程。Java線程啓動時會建立一個本地操做系統線程；當該Java線程終止時，這個操做系統線程也會被回收。操做系統會調度全部線程並將它們分配給可用的CPU。

在上層，Java多線程程序一般把應用分解爲若干個任務，而後使用用戶級的調度器（Executor框架）將這些任務映射爲固定數量的線程；在底層，操做系統內核將這些線程映射到硬件處理器上。

從圖中能夠看出，應用程序經過Executor框架控制上層的調度；而下層的調度由操做系統內核控制，下層的調度不受應用程序的控制。

三大組成部分：任務，任務的執行，異步計算的結果

任務：

包括被執行任務須要實現的接口：Runnable接口或Callable接口。

任務的執行：

包括任務執行機制的核心接口Executor，以及繼承自Executor的ExecutorService接口。Executor框架有兩個關鍵類實現了ExecutorService接口（ThreadPoolExecutor和ScheduledThreadPoolExecutor）。

異步計算的結果：

包括接口Future和實現Future接口的FutureTask類。

成員結構圖

Executor是一個接口，它是Executor框架的基礎，它將任務的提交與任務的執行分離開來。

ExecutorService接口繼承了Executor，在其上作了一些shutdown()、submit()的擴展，能夠說是真正的線程池接口；

AbstractExecutorService抽象類實現了ExecutorService接口中的大部分方法；

ThreadPoolExecutor是線程池的核心實現類，用來執行被提交的任務。

ScheduledExecutorService接口繼承了ExecutorService接口，提供了帶"週期執行"功能ExecutorService；

ScheduledThreadPoolExecutor是一個實現類，能夠在給定的延遲後運行命令，或者按期執行命令。ScheduledThreadPoolExecutor比Timer更靈活，功能更強大。

Future接口和實現Future接口的FutureTask類，表明異步計算的結果。

Runnable接口和Callable接口的實現類，均可以被ThreadPoolExecutor或Scheduled-ThreadPoolExecutor執行。

Executor框架基本使用流程

主線程首先要建立實現Runnable或者Callable接口的任務對象。

工具類Executors能夠把一個Runnable對象封裝爲一個Callable對象（Executors.callable（Runnable task）或Executors.callable（Runnable task，Object resule））。而後能夠把Runnable對象直接交給ExecutorService執行（ExecutorService.execute（Runnablecommand））；或者也能夠把Runnable對象或Callable對象提交給ExecutorService執行（Executor-Service.submit（Runnable task）或ExecutorService.submit（Callable<T>task））。

若是執行ExecutorService.submit（…），ExecutorService將返回一個實現Future接口的對象（到目前爲止的JDK中，返回的是FutureTask對象）。因爲FutureTask實現了Runnable，程序員也能夠建立FutureTask，而後直接交給ExecutorService執行。

最後，主線程能夠執行FutureTask.get()方法來等待任務執行完成。主線程也能夠執行FutureTask.cancel（boolean mayInterruptIfRunning）來取消此任務的執行。

ThreadPoolExecutor一般使用工廠類Executors來建立。Executors能夠建立3種類型的ThreadPoolExecutor：SingleThreadExecutor、FixedThreadPool和CachedThreadPool。

2.Executor框架中的幾個類的理解

FixedThreadPool詳解

1 public static ExecutorService newFixedThreadPool(int nThreads) {
2         return new ThreadPoolExecutor(nThreads, nThreads,
3                                       0L, TimeUnit.MILLISECONDS,
4                                       new LinkedBlockingQueue<Runnable>());
5     }

建立使用固定線程數的FixedThreadPool的API。適用於爲了知足資源管理的需求，而須要限制當前線程數量的應用場景，適用於負載比較重的服務器。FixedThreadPool的corePoolSize和maximumPoolSize都被設置爲建立FixedThreadPool時指定的參數nThreads。

當線程池中的線程數大於corePoolSize時，keepAliveTime爲多餘的空閒線程等待新任務的最長時間，超過這個時間後多餘的線程將被終止。這裏把keepAliveTime設置爲0L，意味着多餘的空閒線程會被當即終止。

FixedThreadPool使用無界隊列LinkedBlockingQueue做爲線程池的工做隊列（隊列的容量爲Integer.MAX_VALUE）。使用無界隊列做爲工做隊列會對線程池帶來以下影響。

1）當線程池中的線程數達到corePoolSize後，新任務將在無界隊列中等待，所以線程池中的線程數不會超過corePoolSize。

2）因爲1，使用無界隊列時maximumPoolSize將是一個無效參數。

3）因爲1和2，使用無界隊列時keepAliveTime將是一個無效參數。

4）因爲使用無界隊列，運行中的FixedThreadPool（未執行方法shutdown()或

shutdownNow()）不會拒絕任務，由於源碼使用的無界阻塞隊列，隊列永遠不會滿（不會調用RejectedExecutionHandler.rejectedExecution方法）。

SingleThreadExecutor詳解

1 public static ExecutorService newSingleThreadExecutor() {
2         return new FinalizableDelegatedExecutorService
3             (new ThreadPoolExecutor(1, 1,
4                                     0L, TimeUnit.MILLISECONDS,
5                                     new LinkedBlockingQueue<Runnable>()));
6     }

建立使用單個線程的SingleThread-Executor的API，適用於須要保證順序地執行各個任務；而且在任意時間點，不會有多個線程是活動的應用場景。

corePoolSize和maximumPoolSize被設置爲1。其餘參數與FixedThreadPool相同。SingleThreadExecutor使用無界隊列LinkedBlockingQueue做爲線程池的工做隊列（隊列的容量爲Integer.MAX_VALUE）。

CachedThreadPool詳解

1 public static ExecutorService newCachedThreadPool() {
2         return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
3                                       60L, TimeUnit.SECONDS,
4                                       new SynchronousQueue<Runnable>());
5     }

建立一個會根據須要建立新線程的CachedThreadPool的API。大小無界的線程池，適用於執行不少的短時間異步任務的小程序，或者是負載較輕的服務器。

corePoolSize被設置爲0，即corePool爲空；maximumPoolSize被設置爲Integer.MAX_VALUE，即maximumPool是無界的。這裏把keepAliveTime設置爲60L，意味着CachedThreadPool中的空閒線程等待新任務的最長時間爲60秒，空閒線程超過60秒後將會被終止。

FixedThreadPool和SingleThreadExecutor使用無界隊列LinkedBlockingQueue做爲線程池的工做隊列。CachedThreadPool使用沒有容量的SynchronousQueue做爲線程池的工做隊列，但CachedThreadPool的maximumPool是無界的。這意味着，若是主線程提交任務的速度高於maximumPool中線程處理任務的速度時，CachedThreadPool會不斷建立新線程。極端狀況下，CachedThreadPool會由於建立過多線程而耗盡CPU和內存資源。

WorkStealingPool

1 public static ExecutorService newWorkStealingPool() {
2         return new ForkJoinPool
3             (Runtime.getRuntime().availableProcessors(),
4              ForkJoinPool.defaultForkJoinWorkerThreadFactory,
5              null, true);
6     }

利用全部運行的處理器數目來建立一個工做竊取的線程池，使用fork/join實現。

ScheduledThreadPoolExecutor詳解

使用工廠類Executors來建立。Executors能夠建立2種類

型的ScheduledThreadPoolExecutor，以下。

·ScheduledThreadPoolExecutor。包含若干個線程的ScheduledThreadPoolExecutor。

·SingleThreadScheduledExecutor。只包含一個線程的ScheduledThreadPoolExecutor。

ScheduledThreadPoolExecutor適用於須要多個後臺線程執行週期任務，同時爲了知足資源管理的需求而須要限制後臺線程的數量的應用場景。

SingleThreadScheduledExecutor適用於須要單個後臺線程執行週期任務，同時須要保證順序地執行各個任務的應用場景。

對這4個步驟的說明。

1）線程1從DelayQueue中獲取已到期的ScheduledFutureTask（DelayQueue.take()）。到期任務是指ScheduledFutureTask的time大於等於當前時間。

2）線程1執行這個ScheduledFutureTask。

3）線程1修改ScheduledFutureTask的time變量爲下次將要被執行的時間。

4）線程1把這個修改time以後的ScheduledFutureTask放回DelayQueue中（Delay-

Queue.add()）。

有關提交定時任務的四個方法：

//向定時任務線程池提交一個延時Runnable任務（僅執行一次）

public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit)

//向定時任務線程池提交一個延時的Callable任務（僅執行一次）

public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit);

//向定時任務線程池提交一個固定時間間隔執行的任務

public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay,

long period, TimeUnit unit)

//向定時任務線程池提交一個固定延時間隔執行的任務

public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay,

long delay, TimeUnit unit);

固定時間間隔的任務不論每次任務花費多少時間，下次任務開始執行時間是肯定的，固然執行任務的時間不能超過執行週期。

固定延時間隔的任務是指每次執行完任務之後都延時一個固定的時間。因爲操做系統調度以及每次任務執行的語句可能不一樣，因此每次任務執行所花費的時間是不肯定的，也就致使了每次任務的執行週期存在必定的波動。

注意：定時或延時任務中所涉及到時間、週期不能保證明時性及準確性，實際運行中會有必定的偏差。

ScheduleThreadPoolExecutor與Timer相比的優點。

（1）Timer是基於絕對時間的延時執行或週期執行，當系統時間改變，則任務的執行會受到的影響。而ScheduleThreadPoolExecutore中，任務時基於相對時間進行週期或延時操做。

（2）Timer也能夠提交多個TimeTask任務，但只有一個線程來執行全部的TimeTask，這樣併發性受到影響。而ScheduleThreadPoolExecutore能夠設定池中線程的數量。

（3）Timer不會捕獲TimerTask的異常，只是簡單地中止，這樣勢必會影響其餘TimeTask的執行。而ScheduleThreadPoolExecutore中，若是一個線程因某些緣由中止，線程池能夠自動建立新的線程來維護池中線程的數量。

scheduleAtFixedRate定時任務超時問題

若任務處理時長超出設置的定時頻率時長,本次任務執行完纔開始下次任務,下次任務已經處於超時狀態,會立刻開始執行.

若任務處理時長小於定時頻率時長,任務執行完後,定時器等待,下次任務會在定時器等待頻率時長後執行

以下例子:

設置定時任務每60s執行一次

若第一次任務時長80s,第二次任務時長20ms,第三次任務時長50ms

第一次任務第0s開始,第80s結束;

第二次任務第80s開始,第110s結束;(上次任務已超時,本次不會再等待60s,會立刻開始),

第三次任務第150s開始,第200s結束.

第四次任務第210s開始.....

ScheduledThreadPoolExecutor實戰：

ScheduleTask.java

1 import java.text.SimpleDateFormat;
 2 import java.util.Date;
 3 
 4 /**
 5  * ScheduledThreadPoolExecutor示例
 6  */
 7 public class ScheduleTask implements Runnable {
 8 
 9     public static enum OperType{
10         None,OnlyThrowException,CatheException
11     }
12 
13     public static SimpleDateFormat formater = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
14 
15     private OperType operType;
16 
17     public ScheduleTask(OperType operType) {
18         this.operType = operType;
19     }
20 
21     @Override
22     public void run() {
23 
24         switch (operType){
25             case OnlyThrowException:
26                 System.out.println("Exception not catch:"+formater.format(new Date()));
27                 throw new RuntimeException("OnlyThrowException");
28             case CatheException:
29                 try {
30                     throw new RuntimeException("CatheException");
31                 } catch (RuntimeException e) {
32                     System.out.println("Exception be catched:"+formater.format(new Date()));
33                 }
34                 break;
35             case None:
36                 System.out.println("None :"+formater.format(new Date()));
37         }
38     }
39 }

TestSchedule.java

1 import java.util.Date;
 2 import java.util.concurrent.ScheduledThreadPoolExecutor;
 3 import java.util.concurrent.TimeUnit;
 4 
 5 /**
 6  * ScheduledThreadPoolExecutor示例測試
 7  */
 8 public class TestSchedule {
 9     public static void main(String[] args) {
10         ScheduledThreadPoolExecutor exec = new ScheduledThreadPoolExecutor(1);
11 
12         /**
13          * 每隔一段時間打印系統時間，互不影響的建立並執行一個在給定初始延遲後首次啓用的按期操做，
14          * 後續操做具備給定的週期；
15          * 也就是將在 initialDelay 後開始執行，週期爲period。
16          */
17         exec.scheduleAtFixedRate(new ScheduleTask(ScheduleTask.OperType.None),
18                 1000,5000, TimeUnit.MILLISECONDS);
19 
20         // 開始執行後就觸發異常,next週期將不會運行
21         exec.scheduleAtFixedRate(new ScheduleTask(ScheduleTask.OperType.OnlyThrowException),
22                 1000,5000, TimeUnit.MILLISECONDS);
23 
24         // 雖然拋出了運行異常,當被攔截了,next週期繼續運行
25         exec.scheduleAtFixedRate(new ScheduleTask(ScheduleTask.OperType.CatheException),
26                 1000,5000, TimeUnit.MILLISECONDS);
27 
28         /**
29          * 建立並執行一個在給定初始延遲後首次啓用的按期操做，
30          * 隨後，在每一次執行終止和下一次執行開始之間都存在給定的延遲。
31          */
32         exec.scheduleWithFixedDelay(new Runnable() {
33             @Override
34             public void run() {
35                 System.out.println("scheduleWithFixedDelay:begin"
36                         +ScheduleTask.formater.format(new Date()));
37                 try {
38                     Thread.sleep(2000);
39                 } catch (InterruptedException e) {
40                     e.printStackTrace();
41                 }
42                 System.out.println("scheduleWithFixedDelay:end"
43                         +ScheduleTask.formater.format(new Date()));
44             }
45         },1000,5000,TimeUnit.MILLISECONDS);
46 
47         /**
48          * 建立並執行在給定延遲後啓用的一次性操做。
49          */
50         exec.schedule(new Runnable() {
51             @Override
52             public void run() {
53                 System.out.println("schedule running.....");
54             }
55         },5000,TimeUnit.MILLISECONDS);
56     }
57 
58 
59 
60 }

Callable、Future和FutureTask詳解

Future接口和實現Future接口的FutureTask類用來表示異步計算的結果。

當咱們把Runnable接口或Callable接口的實現類提交（submit）給ThreadPoolExecutor或ScheduledThreadPoolExecutor時，ThreadPoolExecutor或ScheduledThreadPoolExecutor會向咱們返回一個FutureTask對象。

Runnable接口和Callable接口的實現類，均可以被ThreadPoolExecutor或Scheduled-ThreadPoolExecutor執行。它們之間的區別是Runnable不會返回結果，而Callable能夠返回結果。

除了能夠本身建立實現Callable接口的對象外，還可使用工廠類Executors來把一個Runnable包裝成一個Callable。

Executors提供的，把一個Runnable包裝成一個Callable的API。

public static Callable<Object> callable(Runnable task) // 假設返回對象Callable1

Executors提供的，把一個Runnable和一個待返回的結果包裝成一個Callable的API。

public static <T> Callable<T> callable(Runnable task, T result) // 假設返回對象Callable2

當任務成功完成後FutureTask.get()將返回該任務的結果。例如，若是提交的是對象Callable1，FutureTask.get()方法將返回null；若是提交的是對象Callable2，FutureTask.get()方法將返回result對象。

FutureTask除了實現Future接口外，還實現了Runnable接口。所以，FutureTask能夠交給Executor執行，也能夠由調用線程直接執行（FutureTask.run()）。

當FutureTask處於未啓動或已啓動狀態時，執行FutureTask.get()方法將致使調用線程阻塞；當FutureTask處於已完成狀態時，執行FutureTask.get()方法將致使調用線程當即返回結果或拋出異常。

當FutureTask處於未啓動狀態時，執行FutureTask.cancel()方法將致使此任務永遠不會被執行；當FutureTask處於已啓動狀態時，執行FutureTask.cancel（true）方法將以中斷執行此任務線程的方式來試圖中止任務；當FutureTask處於已啓動狀態時，執行FutureTask.cancel（false）方法將不會對正在執行此任務的線程產生影響（讓正在執行的任務運行完成）；當FutureTask處於已完成狀態時，執行FutureTask.cancel（…）方法將返回false。

實戰：

ComputeTask.java

1 import java.util.concurrent.Callable;
 2 
 3 /**
 4  * lgs
 5  * 建立日期：2017/12/10
 6  * 建立時間: 21:38
 7  */
 8 public class ComputeTask implements Callable<Integer> {
 9 
10     private Integer result =0;
11     private String taskName ="";
12 
13     public ComputeTask(Integer result, String taskName) {
14         this.result = result;
15         this.taskName = taskName;
16         System.out.println(taskName+"子任務已經建立");
17     }
18 
19     @Override
20     public Integer call() throws Exception {
21         for(int i=0;i<100;i++){
22             result = result+i;
23         }
24         Thread.sleep(2000);
25         System.out.println(taskName+"子任務已經完成");
26         return  result;
27     }
28 }

FutureSample.java

1 import java.util.ArrayList;
 2 import java.util.List;
 3 import java.util.concurrent.*;
 4 
 5 /**
 6  * lgs 建立日期：2017/12/10 建立時間: 21:41
 7  */
 8 public class FutureSample {
 9     public static void main(String[] args) {
10         FutureSample futureSample = new FutureSample();
11         // 建立任務集合
12         List<FutureTask<Integer>> taskList = new ArrayList<>();
13         // 另外一種方式
14         // List<Future<Integer>> futureList = new ArrayList<>();
15         ExecutorService exec = Executors.newFixedThreadPool(5);
16         for (int i = 0; i < 10; i++) {
17             // 傳入Callable對象建立FutureTask對象
18             FutureTask<Integer> ft = new FutureTask<Integer>(new ComputeTask(i, "task_" + i));
19             taskList.add(ft);
20             exec.submit(ft);
21             // Future<Integer> result = exec.submit(new ComputeTask(i,"task_"+i));
22             // futureList.add(result);
23         }
24         System.out.println("主線程已經提交任務，作本身的事！");
25 
26         // 開始統計各計算線程計算結果
27         int totalResult = 0;
28         for (FutureTask<Integer> ft : taskList) {
29             try {
30                 // FutureTask的get方法會自動阻塞,直到獲取計算結果爲止
31                 totalResult = totalResult + ft.get();
32             } catch (InterruptedException e) {
33                 e.printStackTrace();
34             } catch (ExecutionException e) {
35                 e.printStackTrace();
36             }
37         }
38         System.out.println("total = " + totalResult);
39         exec.shutdown();
40 
41     }
42 }

CompletionService詳解

CompletionService實際上能夠看作是Executor和BlockingQueue的結合體。CompletionService在接收到要執行的任務時，經過相似BlockingQueue的put和take得到任務執行的結果。CompletionService的一個實現是ExecutorCompletionService，ExecutorCompletionService把具體的計算任務交給Executor完成。在實現上，ExecutorCompletionService在構造函數中會建立一個BlockingQueue（使用的基於鏈表的無界隊列LinkedBlockingQueue），該BlockingQueue的做用是保存Executor執行的結果。當計算完成時，調用FutureTask的done方法。當提交一個任務到ExecutorCompletionService時，首先將任務包裝成QueueingFuture，它是FutureTask的一個子類，而後改寫FutureTask的done方法，以後把Executor執行的計算結果放入BlockingQueue中。

與ExecutorService最主要的區別在於submit的task不必定是按照加入時的順序完成的。CompletionService對ExecutorService進行了包裝，內部維護一個保存Future對象的BlockingQueue。只有當這個Future對象狀態是結束的時候，纔會加入到這個Queue中，take()方法其實就是Producer-Consumer中的Consumer。它會從Queue中取出Future對象，若是Queue是空的，就會阻塞在那裏，直到有完成的Future對象加入到Queue中。因此，先完成的一定先被取出。這樣就減小了沒必要要的等待時間。

總結：

使用方法一，本身建立一個集合來保存Future存根並循環調用其返回結果的時候，主線程並不能保證首先得到的是最早完成任務的線程返回值。它只是按加入線程池的順序返回。由於take方法是阻塞方法，後面的任務完成了，前面的任務卻沒有完成，主程序就那樣等待在那兒，只到前面的完成了，它才知道原來後面的也完成了。

使用方法二，使用CompletionService來維護處理線程不的返回結果時，主線程老是可以拿到最早完成的任務的返回值，而無論它們加入線程池的順序。

WorkTask.java

1 import java.util.Random;
 2 import java.util.concurrent.Callable;
 3 
 4 /**
 5  * CompletionService的任務類
 6  */
 7 public class WorkTask implements Callable<String> {
 8 
 9     private String name;
10 
11     public WorkTask(String name) {
12         this.name = name;
13     }
14 
15     @Override
16     public String call() throws Exception {
17         //休眠隨機時間，觀察獲取結果的行爲。
18         int sleepTime = new Random().nextInt(1000);
19         Thread.sleep(sleepTime);
20         String str = name+" sleept time:"+sleepTime;
21         System.out.println(str+" finished....");
22         return str;
23     }
24 }

CompletionTest.java

1 import java.util.concurrent.*;
 2 
 3 /**
 4  * CompletionService測試類
 5  */
 6 public class CompletionTest {
 7     private final int POOL_SIZE  = 5;
 8     private final int TOTAL_TASK = 10;
 9 
10     // 方法一，本身寫集合來實現獲取線程池中任務的返回結果
11     public void testByQueue() throws InterruptedException, ExecutionException {
12         ExecutorService pool = Executors.newFixedThreadPool(POOL_SIZE);
13         BlockingQueue<Future<String>> queue = new LinkedBlockingDeque<>();
14         // 向裏面扔任務
15         for(int i=0;i<TOTAL_TASK;i++){
16             Future<String> future = pool.submit(new WorkTask("ExecTask"+i));
17             queue.add(future);
18         }
19 
20         // 檢查線程池任務執行結果
21         for(int i=0;i<TOTAL_TASK;i++){
22             System.out.println("ExecTask:"+queue.take().get());
23         }
24 
25         pool.shutdown();
26 
27     }
28 
29     // 方法二，經過CompletionService來實現獲取線程池中任務的返回結果
30     public void testByCompletion() throws InterruptedException, ExecutionException {
31         ExecutorService pool = Executors.newFixedThreadPool(POOL_SIZE);
32         CompletionService<String> service = new ExecutorCompletionService<String>(pool);
33 
34         // 向裏面扔任務
35         for(int i=0;i<TOTAL_TASK;i++){
36             service.submit(new WorkTask("ExecTask"+i) );
37         }
38 
39         // 檢查線程池任務執行結果
40         for(int i=0;i<TOTAL_TASK;i++){
41             Future<String> future = service.take();
42             System.out.println("CompletionService:"+future.get());
43         }
44 
45     }
46 
47     public static void main(String[] args) throws ExecutionException, InterruptedException {
48         CompletionTest  completionTest = new CompletionTest();
49         //completionTest.testByQueue();
50         completionTest.testByCompletion();
51     }
52 
53 }