线程池的优点
線程池的優(yōu)點
1、線程是稀缺資源，使用線程池可以減少創(chuàng)建和銷毀線程的次數(shù)，每個工作線程都可以重復(fù)使用。
2、可以根據(jù)系統(tǒng)的承受能力，調(diào)整線程池中工作線程的數(shù)量，防止因為消耗過多內(nèi)存導(dǎo)致服務(wù)器崩潰。
線程池的創(chuàng)建
public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue,RejectedExecutionHandler handler)corePoolSize：線程池核心線程數(shù)量
maximumPoolSize:線程池最大線程數(shù)量
keepAliverTime：當(dāng)活躍線程數(shù)大于核心線程數(shù)時，空閑的多余線程最大存活時間
unit：存活時間的單位
workQueue：存放任務(wù)的隊列
handler：超出線程范圍和隊列容量的任務(wù)的處理程序
線程池的實現(xiàn)原理
提交一個任務(wù)到線程池中，線程池的處理流程如下：
1、判斷線程池里的核心線程是否都在執(zhí)行任務(wù)，如果不是（核心線程空閑或者還有核心線程沒有被創(chuàng)建）則創(chuàng)建一個新的工作線程來執(zhí)行任務(wù)。如果核心線程都在執(zhí)行任務(wù)，則進(jìn)入下個流程。
2、線程池判斷工作隊列是否已滿，如果工作隊列沒有滿，則將新提交的任務(wù)存儲在這個工作隊列里。如果工作隊列滿了，則進(jìn)入下個流程。
3、判斷線程池里的線程是否都處于工作狀態(tài)，如果沒有，則創(chuàng)建一個新的工作線程來執(zhí)行任務(wù)。如果已經(jīng)滿了，則交給飽和策略來處理這個任務(wù)。
??
線程池的源碼解讀
1、ThreadPoolExecutor的execute()方法
?
1 public void execute(Runnable command) {2 if (command == null)3 throw new NullPointerException();//如果線程數(shù)大于等于基本線程數(shù)或者線程創(chuàng)建失敗，將任務(wù)加入隊列4 if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {//線程池處于運(yùn)行狀態(tài)并且加入隊列成功5 if (runState == RUNNING && workQueue.offer(command)) {6 if (runState != RUNNING || poolSize == 0)7 ensureQueuedTaskHandled(command);8 }//線程池不處于運(yùn)行狀態(tài)或者加入隊列失敗，則創(chuàng)建線程（創(chuàng)建的是非核心線程）9 else if (!addIfUnderMaximumPoolSize(command))//創(chuàng)建線程失敗，則采取阻塞處理的方式 10 reject(command); // is shutdown or saturated 11 } 12 }?
2、創(chuàng)建線程的方法：addIfUnderCorePoolSize(command)
?
1 private boolean addIfUnderCorePoolSize(Runnable firstTask) {2 Thread t = null;3 final ReentrantLock mainLock = this.mainLock;4 mainLock.lock();5 try {6 if (poolSize < corePoolSize && runState == RUNNING)7 t = addThread(firstTask);8 } finally {9 mainLock.unlock(); 10 } 11 if (t == null) 12 return false; 13 t.start(); 14 return true; 15 }?
我們重點來看第7行：
?
1 private Thread addThread(Runnable firstTask) {2 Worker w = new Worker(firstTask);3 Thread t = threadFactory.newThread(w);4 if (t != null) {5 w.thread = t;6 workers.add(w);7 int nt = ++poolSize;8 if (nt > largestPoolSize)9 largestPoolSize = nt; 10 } 11 return t; 12 }?
這里將線程封裝成工作線程worker，并放入工作線程組里，worker類的方法run方法：
?
public void run() {try {Runnable task = firstTask;firstTask = null;while (task != null || (task = getTask()) != null) {runTask(task);task = null;}} finally {workerDone(this);}}?
worker在執(zhí)行完任務(wù)后，還會通過getTask方法循環(huán)獲取工作隊里里的任務(wù)來執(zhí)行。
我們通過一個程序來觀察線程池的工作原理：
1、創(chuàng)建一個線程
?
1 public class ThreadPoolTest implements Runnable2 {3 @Override4 public void run()5 {6 try7 {8 Thread.sleep(300);9 } 10 catch (InterruptedException e) 11 { 12 e.printStackTrace(); 13 } 14 } 15 }?
2、線程池循環(huán)運(yùn)行16個線程：
?
1 public static void main(String[] args)2 {3 LinkedBlockingQueue<Runnable> queue =4 new LinkedBlockingQueue<Runnable>(5);5 ThreadPoolExecutor threadPool = new ThreadPoolExecutor(5, 10, 60, TimeUnit.SECONDS, queue);6 for (int i = 0; i < 16 ; i++)7 {8 threadPool.execute(9 new Thread(new ThreadPoolTest(), "Thread".concat(i + ""))); 10 System.out.println("線程池中活躍的線程數(shù)： " + threadPool.getPoolSize()); 11 if (queue.size() > 0) 12 { 13 System.out.println("----------------隊列中阻塞的線程數(shù)" + queue.size()); 14 } 15 } 16 threadPool.shutdown(); 17 }?
執(zhí)行結(jié)果：
?
線程池中活躍的線程數(shù)： 1 線程池中活躍的線程數(shù)： 2 線程池中活躍的線程數(shù)： 3 線程池中活躍的線程數(shù)： 4 線程池中活躍的線程數(shù)： 5 線程池中活躍的線程數(shù)： 5 ----------------隊列中阻塞的線程數(shù)1 線程池中活躍的線程數(shù)： 5 ----------------隊列中阻塞的線程數(shù)2 線程池中活躍的線程數(shù)： 5 ----------------隊列中阻塞的線程數(shù)3 線程池中活躍的線程數(shù)： 5 ----------------隊列中阻塞的線程數(shù)4 線程池中活躍的線程數(shù)： 5 ----------------隊列中阻塞的線程數(shù)5 線程池中活躍的線程數(shù)： 6 ----------------隊列中阻塞的線程數(shù)5 線程池中活躍的線程數(shù)： 7 ----------------隊列中阻塞的線程數(shù)5 線程池中活躍的線程數(shù)： 8 ----------------隊列中阻塞的線程數(shù)5 線程池中活躍的線程數(shù)： 9 ----------------隊列中阻塞的線程數(shù)5 線程池中活躍的線程數(shù)： 10 ----------------隊列中阻塞的線程數(shù)5 Exception in thread "main" java.util.concurrent.RejectedExecutionException: Task Thread[Thread15,5,main] rejected from java.util.concurrent.ThreadPoolExecutor@232204a1[Running, pool size = 10, active threads = 10, queued tasks = 5, completed tasks = 0]at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2047)at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:823)at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1369)at test.ThreadTest.main(ThreadTest.java:17)?
從結(jié)果可以觀察出：
1、創(chuàng)建的線程池具體配置為：核心線程數(shù)量為5個；全部線程數(shù)量為10個；工作隊列的長度為5。
2、我們通過queue.size()的方法來獲取工作隊列中的任務(wù)數(shù)。
3、運(yùn)行原理：
??????剛開始都是在創(chuàng)建新的線程，達(dá)到核心線程數(shù)量5個后，新的任務(wù)進(jìn)來后不再創(chuàng)建新的線程，而是將任務(wù)加入工作隊列，任務(wù)隊列到達(dá)上線5個后，新的任務(wù)又會創(chuàng)建新的普通線程，直到達(dá)到線程池最大的線程數(shù)量10個，后面的任務(wù)則根據(jù)配置的飽和策略來處理。我們這里沒有具體配置，使用的是默認(rèn)的配置AbortPolicy:直接拋出異常。
當(dāng)然，為了達(dá)到我需要的效果，上述線程處理的任務(wù)都是利用休眠導(dǎo)致線程沒有釋放！！！
RejectedExecutionHandler：飽和策略
當(dāng)隊列和線程池都滿了，說明線程池處于飽和狀態(tài)，那么必須對新提交的任務(wù)采用一種特殊的策略來進(jìn)行處理。這個策略默認(rèn)配置是AbortPolicy，表示無法處理新的任務(wù)而拋出異常。JAVA提供了4中策略：
1、AbortPolicy：直接拋出異常
2、CallerRunsPolicy：只用調(diào)用所在的線程運(yùn)行任務(wù)
3、DiscardOldestPolicy：丟棄隊列里最近的一個任務(wù)，并執(zhí)行當(dāng)前任務(wù)。
4、DiscardPolicy：不處理，丟棄掉。
我們現(xiàn)在用第四種策略來處理上面的程序：
?
?
1 public static void main(String[] args)2 {3 LinkedBlockingQueue<Runnable> queue =4 new LinkedBlockingQueue<Runnable>(3);5 RejectedExecutionHandler handler = new ThreadPoolExecutor.DiscardPolicy();6 7 ThreadPoolExecutor threadPool = new ThreadPoolExecutor(2, 5, 60, TimeUnit.SECONDS, queue,handler);8 for (int i = 0; i < 9 ; i++)9 { 10 threadPool.execute( 11 new Thread(new ThreadPoolTest(), "Thread".concat(i + ""))); 12 System.out.println("線程池中活躍的線程數(shù)： " + threadPool.getPoolSize()); 13 if (queue.size() > 0) 14 { 15 System.out.println("----------------隊列中阻塞的線程數(shù)" + queue.size()); 16 } 17 } 18 threadPool.shutdown(); 19 }?
?
執(zhí)行結(jié)果：
?
線程池中活躍的線程數(shù)： 1 線程池中活躍的線程數(shù)： 2 線程池中活躍的線程數(shù)： 2 ----------------隊列中阻塞的線程數(shù)1 線程池中活躍的線程數(shù)： 2 ----------------隊列中阻塞的線程數(shù)2 線程池中活躍的線程數(shù)： 2 ----------------隊列中阻塞的線程數(shù)3 線程池中活躍的線程數(shù)： 3 ----------------隊列中阻塞的線程數(shù)3 線程池中活躍的線程數(shù)： 4 ----------------隊列中阻塞的線程數(shù)3 線程池中活躍的線程數(shù)： 5 ----------------隊列中阻塞的線程數(shù)3 線程池中活躍的線程數(shù)： 5 ----------------隊列中阻塞的線程數(shù)3這里采用了丟棄策略后，就沒有再拋出異常，而是直接丟棄。在某些重要的場景下，可以采用記錄日志或者存儲到數(shù)據(jù)庫中，而不應(yīng)該直接丟棄。
設(shè)置策略有兩種方式：
1、
RejectedExecutionHandler handler = new ThreadPoolExecutor.DiscardPolicy();ThreadPoolExecutor threadPool = new ThreadPoolExecutor(2, 5, 60, TimeUnit.SECONDS, queue,handler);2、
ThreadPoolExecutor threadPool = new ThreadPoolExecutor(2, 5, 60, TimeUnit.SECONDS, queue);threadPool.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy());Executor框架的兩級調(diào)度模型
在HotSpot VM的模型中，JAVA線程被一對一映射為本地操作系統(tǒng)線程。JAVA線程啟動時會創(chuàng)建一個本地操作系統(tǒng)線程，當(dāng)JAVA線程終止時，對應(yīng)的操作系統(tǒng)線程也被銷毀回收，而操作系統(tǒng)會調(diào)度所有線程并將它們分配給可用的CPU。
在上層，JAVA程序會將應(yīng)用分解為多個任務(wù)，然后使用應(yīng)用級的調(diào)度器（Executor）將這些任務(wù)映射成固定數(shù)量的線程；在底層，操作系統(tǒng)內(nèi)核將這些線程映射到硬件處理器上。
Executor框架類圖
在前面介紹的JAVA線程既是工作單元，也是執(zhí)行機(jī)制。而在Executor框架中，我們將工作單元與執(zhí)行機(jī)制分離開來。Runnable和Callable是工作單元（也就是俗稱的任務(wù)），而執(zhí)行機(jī)制由Executor來提供。這樣一來Executor是基于生產(chǎn)者消費(fèi)者模式的，提交任務(wù)的操作相當(dāng)于生成者，執(zhí)行任務(wù)的線程相當(dāng)于消費(fèi)者。
1、從類圖上看，Executor接口是異步任務(wù)執(zhí)行框架的基礎(chǔ)，該框架能夠支持多種不同類型的任務(wù)執(zhí)行策略。
public interface Executor {void execute(Runnable command); }Executor接口就提供了一個執(zhí)行方法，任務(wù)是Runnbale類型，不支持Callable類型。
2、ExecutorService接口實現(xiàn)了Executor接口，主要提供了關(guān)閉線程池和submit方法：
?
public interface ExecutorService extends Executor {List<Runnable> shutdownNow();boolean isTerminated();<T> Future<T> submit(Callable<T> task);}?
另外該接口有兩個重要的實現(xiàn)類：ThreadPoolExecutor與ScheduledThreadPoolExecutor。
其中ThreadPoolExecutor是線程池的核心實現(xiàn)類，用來執(zhí)行被提交的任務(wù)；而ScheduledThreadPoolExecutor是一個實現(xiàn)類，可以在給定的延遲后運(yùn)行任務(wù)，或者定期執(zhí)行命令。
在上一篇文章中，我是使用ThreadPoolExecutor來通過給定不同的參數(shù)從而創(chuàng)建自己所需的線程池，但是在后面的工作中不建議這種方式，推薦使用Exectuors工廠方法來創(chuàng)建線程池
這里先來區(qū)別線程池和線程組（ThreadGroup與ThreadPoolExecutor）這兩個概念：
a、線程組就表示一個線程的集合。
b、線程池是為線程的生命周期開銷問題和資源不足問題提供解決方案，主要是用來管理線程。
Executors可以創(chuàng)建3種類型的ThreadPoolExecutor：SingleThreadExecutor、FixedThreadExecutor和CachedThreadPool
a、SingleThreadExecutor：單線程線程池
ExecutorService threadPool = Executors.newSingleThreadExecutor(); public static ExecutorService newSingleThreadExecutor() {return new FinalizableDelegatedExecutorService(new ThreadPoolExecutor(1, 1,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>()));}我們從源碼來看可以知道，單線程線程池的創(chuàng)建也是通過ThreadPoolExecutor，里面的核心線程數(shù)和線程數(shù)都是1，并且工作隊列使用的是無界隊列。由于是單線程工作，每次只能處理一個任務(wù)，所以后面所有的任務(wù)都被阻塞在工作隊列中，只能一個個任務(wù)執(zhí)行。
b、FixedThreadExecutor：固定大小線程池
ExecutorService threadPool = Executors.newFixedThreadPool(5); public static ExecutorService newFixedThreadPool(int nThreads) {return new ThreadPoolExecutor(nThreads, nThreads,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>());}這個與單線程類似，只是創(chuàng)建了固定大小的線程數(shù)量。
c、CachedThreadPool:無界線程池
ExecutorService threadPool = Executors.newCachedThreadPool(); public static ExecutorService newCachedThreadPool() {return new ThreadPoolExecutor(0, Integer.MAX_VALUE,60L, TimeUnit.SECONDS,new SynchronousQueue<Runnable>());}無界線程池意味著沒有工作隊列，任務(wù)進(jìn)來就執(zhí)行，線程數(shù)量不夠就創(chuàng)建，與前面兩個的區(qū)別是：空閑的線程會被回收掉，空閑的時間是60s。這個適用于執(zhí)行很多短期異步的小程序或者負(fù)載較輕的服務(wù)器。
Callable、Future、FutureTash詳解
Callable與Future是在JAVA的后續(xù)版本中引入進(jìn)來的，Callable類似于Runnable接口，實現(xiàn)Callable接口的類與實現(xiàn)Runnable的類都是可以被線程執(zhí)行的任務(wù)。
三者之間的關(guān)系：
Callable是Runnable封裝的異步運(yùn)算任務(wù)。
Future用來保存Callable異步運(yùn)算的結(jié)果
FutureTask封裝Future的實體類
1、Callable與Runnbale的區(qū)別
a、Callable定義的方法是call，而Runnable定義的方法是run。
b、call方法有返回值，而run方法是沒有返回值的。
c、call方法可以拋出異常，而run方法不能拋出異常。
2、Future
Future表示異步計算的結(jié)果，提供了以下方法，主要是判斷任務(wù)是否完成、中斷任務(wù)、獲取任務(wù)執(zhí)行結(jié)果
?
1 public interface Future<V> {2 3 boolean cancel(boolean mayInterruptIfRunning);4 5 boolean isCancelled();6 7 boolean isDone();8 9 V get() throws InterruptedException, ExecutionException; 10 11 V get(long timeout, TimeUnit unit) 12 throws InterruptedException, ExecutionException, TimeoutException; 13 }?
3、FutureTask<V>
可取消的異步計算，此類提供了對Future的基本實現(xiàn)，僅在計算完成時才能獲取結(jié)果，如果計算尚未完成，則阻塞get方法。
public class FutureTask<V> implements RunnableFuture<V> public interface RunnableFuture<V> extends Runnable, Future<V>FutureTask不僅實現(xiàn)了Future接口，還實現(xiàn)了Runnable接口，所以不僅可以將FutureTask當(dāng)成一個任務(wù)交給Executor來執(zhí)行，還可以通過Thread來創(chuàng)建一個線程。
Callable與FutureTask
定義一個callable的任務(wù)：
?
1 public class MyCallableTask implements Callable<Integer>2 {3 @Override4 public Integer call()5 throws Exception6 {7 System.out.println("callable do somothing");8 Thread.sleep(5000);9 return new Random().nextInt(100); 10 } 11 }?
?
1 public class CallableTest2 {3 public static void main(String[] args) throws Exception4 {5 Callable<Integer> callable = new MyCallableTask();6 FutureTask<Integer> future = new FutureTask<Integer>(callable);7 Thread thread = new Thread(future);8 thread.start();9 Thread.sleep(100); 10 //嘗試取消對此任務(wù)的執(zhí)行 11 future.cancel(true); 12 //判斷是否在任務(wù)正常完成前取消 13 System.out.println("future is cancel:" + future.isCancelled()); 14 if(!future.isCancelled()) 15 { 16 System.out.println("future is cancelled"); 17 } 18 //判斷任務(wù)是否已完成 19 System.out.println("future is done:" + future.isDone()); 20 if(!future.isDone()) 21 { 22 System.out.println("future get=" + future.get()); 23 } 24 else 25 { 26 //任務(wù)已完成 27 System.out.println("task is done"); 28 } 29 } 30 }?
執(zhí)行結(jié)果：
callable do somothing future is cancel:true future is done:true task is done這個DEMO主要是通過調(diào)用FutureTask的狀態(tài)設(shè)置的方法，演示了狀態(tài)的變遷。
a、第11行，嘗試取消對任務(wù)的執(zhí)行，該方法如果由于任務(wù)已完成、已取消則返回false，如果能夠取消還未完成的任務(wù)，則返回true，該DEMO中由于任務(wù)還在休眠狀態(tài)，所以可以取消成功。
future.cancel(true);b、第13行，判斷任務(wù)取消是否成功：如果在任務(wù)正常完成前將其取消，則返回true
System.out.println("future is cancel:" + future.isCancelled());c、第19行，判斷任務(wù)是否完成：如果任務(wù)完成，則返回true，以下幾種情況都屬于任務(wù)完成：正常終止、異常或者取消而完成。
??? 我們的DEMO中，任務(wù)是由于取消而導(dǎo)致完成。
System.out.println("future is done:" + future.isDone());d、在第22行，獲取異步線程執(zhí)行的結(jié)果，我這個DEMO中沒有執(zhí)行到這里，需要注意的是，future.get方法會阻塞當(dāng)前線程， 直到任務(wù)執(zhí)行完成返回結(jié)果為止。
System.out.println("future get=" + future.get());Callable與Future
?
public class CallableThread implements Callable<String> {@Overridepublic String call()throws Exception{System.out.println("進(jìn)入Call方法，開始休眠，休眠時間為：" + System.currentTimeMillis());Thread.sleep(10000);return "今天停電";}public static void main(String[] args) throws Exception{ExecutorService es = Executors.newSingleThreadExecutor();Callable<String> call = new CallableThread();Future<String> fu = es.submit(call);es.shutdown();Thread.sleep(5000);System.out.println("主線程休眠5秒，當(dāng)前時間" + System.currentTimeMillis());String str = fu.get();System.out.println("Future已拿到數(shù)據(jù)，str=" + str + ";當(dāng)前時間為：" + System.currentTimeMillis());} }?
執(zhí)行結(jié)果：
進(jìn)入Call方法，開始休眠，休眠時間為：1478606602676 主線程休眠5秒，當(dāng)前時間1478606608676 Future已拿到數(shù)據(jù)，str=今天停電;當(dāng)前時間為：1478606612677這里的future是直接扔到線程池里面去執(zhí)行的。由于要打印任務(wù)的執(zhí)行結(jié)果，所以從執(zhí)行結(jié)果來看，主線程雖然休眠了5s，但是從Call方法執(zhí)行到拿到任務(wù)的結(jié)果，這中間的時間差正好是10s，說明get方法會阻塞當(dāng)前線程直到任務(wù)完成。
通過FutureTask也可以達(dá)到同樣的效果：
?
?
public static void main(String[] args) throws Exception{ExecutorService es = Executors.newSingleThreadExecutor();Callable<String> call = new CallableThread();FutureTask<String> task = new FutureTask<String>(call);es.submit(task);es.shutdown();Thread.sleep(5000);System.out.println("主線程等待5秒，當(dāng)前時間為：" + System.currentTimeMillis());String str = task.get();System.out.println("Future已拿到數(shù)據(jù)，str=" + str + ";當(dāng)前時間為：" + System.currentTimeMillis());}?
以上的組合可以給我們帶來這樣的一些變化：
如有一種場景中，方法A返回一個數(shù)據(jù)需要10s,A方法后面的代碼運(yùn)行需要20s，但是這20s的執(zhí)行過程中，只有后面10s依賴于方法A執(zhí)行的結(jié)果。如果與以往一樣采用同步的方式，勢必會有10s的時間被浪費(fèi)，如果采用前面兩種組合，則效率會提高：
1、先把A方法的內(nèi)容放到Callable實現(xiàn)類的call()方法中
2、在主線程中通過線程池執(zhí)行A任務(wù)
3、執(zhí)行后面方法中10秒不依賴方法A運(yùn)行結(jié)果的代碼
4、獲取方法A的運(yùn)行結(jié)果，執(zhí)行后面方法中10秒依賴方法A運(yùn)行結(jié)果的代碼
這樣代碼執(zhí)行效率一下子就提高了，程序不必卡在A方法處。
總結(jié)
- 上一篇： Java多线程同步机制
- 下一篇： linux运维、架构之路-jumpser