轉載自：https://www.cnblogs.com/dolphin0520/p/3932921.html Java併發編程：線程池的使用 在前面的文章中，我們使用線程的時候就去創建一個線程，這樣實現起來非常簡便，但是就會有一個問題： 如果併發的線程數量很多，並且每個線程都是執行一個時間很短的任務 ...
轉載自：https://www.cnblogs.com/dolphin0520/p/3932921.html
Java併發編程：線程池的使用
在前面的文章中，我們使用線程的時候就去創建一個線程，這樣實現起來非常簡便，但是就會有一個問題：
如果併發的線程數量很多，並且每個線程都是執行一個時間很短的任務就結束了，這樣頻繁創建線程就會大大降低系統的效率，因為頻繁創建線程和銷毀線程需要時間。
那麼有沒有一種辦法使得線程可以復用，就是執行完一個任務，並不被銷毀，而是可以繼續執行其他的任務？
在Java中可以通過線程池來達到這樣的效果。今天我們就來詳細講解一下Java的線程池，首先我們從最核心的ThreadPoolExecutor類中的方法講起，然後再講述它的實現原理，接著給出了它的使用示例，最後討論了一下如何合理配置線程池的大小。
以下是本文的目錄大綱：
一.Java中的ThreadPoolExecutor類
二.深入剖析線程池實現原理
三.使用示例
四.如何合理配置線程池的大小
若有不正之處請多多諒解，並歡迎批評指正。
請尊重作者勞動成果，轉載請標明原文鏈接：
http://www.cnblogs.com/dolphin0520/p/3932921.html
一.Java中的ThreadPoolExecutor類
java.uitl.concurrent.ThreadPoolExecutor類是線程池中最核心的一個類，因此如果要透徹地瞭解Java中的線程池，必須先瞭解這個類。下麵我們來看一下ThreadPoolExecutor類的具體實現源碼。
在ThreadPoolExecutor類中提供了四個構造方法：
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 |
public class ThreadPoolExecutor extends AbstractExecutorService {
.....
public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,
BlockingQueue<Runnable> workQueue);
public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,
BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory);
public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,
BlockingQueue<Runnable> workQueue,RejectedExecutionHandler handler);
public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,
BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler);
...
}
|
從上面的代碼可以得知，ThreadPoolExecutor繼承了AbstractExecutorService類，並提供了四個構造器，事實上，通過觀察每個構造器的源碼具體實現，發現前面三個構造器都是調用的第四個構造器進行的初始化工作。
下麵解釋下一下構造器中各個參數的含義：
- corePoolSize：核心池的大小，這個參數跟後面講述的線程池的實現原理有非常大的關係。在創建了線程池後，預設情況下，線程池中並沒有任何線程，而是等待有任務到來才創建線程去執行任務，除非調用了prestartAllCoreThreads()或者prestartCoreThread()方法，從這2個方法的名字就可以看出，是預創建線程的意思，即在沒有任務到來之前就創建corePoolSize個線程或者一個線程。預設情況下，在創建了線程池後，線程池中的線程數為0，當有任務來之後，就會創建一個線程去執行任務，當線程池中的線程數目達到corePoolSize後，就會把到達的任務放到緩存隊列當中；
- maximumPoolSize：線程池最大線程數，這個參數也是一個非常重要的參數，它表示線上程池中最多能創建多少個線程；
- keepAliveTime：表示線程沒有任務執行時最多保持多久時間會終止。預設情況下，只有當線程池中的線程數大於corePoolSize時，keepAliveTime才會起作用，直到線程池中的線程數不大於corePoolSize，即當線程池中的線程數大於corePoolSize時，如果一個線程空閑的時間達到keepAliveTime，則會終止，直到線程池中的線程數不超過corePoolSize。但是如果調用了allowCoreThreadTimeOut(boolean)方法，線上程池中的線程數不大於corePoolSize時，keepAliveTime參數也會起作用，直到線程池中的線程數為0；
- unit：參數keepAliveTime的時間單位，有7種取值，在TimeUnit類中有7種靜態屬性：
TimeUnit.DAYS; //天 TimeUnit.HOURS; //小時 TimeUnit.MINUTES; //分鐘 TimeUnit.SECONDS; //秒 TimeUnit.MILLISECONDS; //毫秒 TimeUnit.MICROSECONDS; //微妙 TimeUnit.NANOSECONDS; //納秒
- workQueue：一個阻塞隊列，用來存儲等待執行的任務，這個參數的選擇也很重要，會對線程池的運行過程產生重大影響，一般來說，這裡的阻塞隊列有以下幾種選擇：
ArrayBlockingQueue; LinkedBlockingQueue; SynchronousQueue;
ArrayBlockingQueue和PriorityBlockingQueue使用較少，一般使用LinkedBlockingQueue和Synchronous。線程池的排隊策略與BlockingQueue有關。
- threadFactory：線程工廠，主要用來創建線程；
- handler：表示當拒絕處理任務時的策略，有以下四種取值：
ThreadPoolExecutor.AbortPolicy:丟棄任務並拋出RejectedExecutionException異常。 ThreadPoolExecutor.DiscardPolicy：也是丟棄任務，但是不拋出異常。 ThreadPoolExecutor.DiscardOldestPolicy：丟棄隊列最前面的任務，然後重新嘗試執行任務（重覆此過程） ThreadPoolExecutor.CallerRunsPolicy：由調用線程處理該任務
具體參數的配置與線程池的關係將在下一節講述。
從上面給出的ThreadPoolExecutor類的代碼可以知道，ThreadPoolExecutor繼承了AbstractExecutorService，我們來看一下AbstractExecutorService的實現：
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 |
public abstract class AbstractExecutorService implements ExecutorService {
protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) { };
protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) { };
public Future<?> submit(Runnable task) {};
public <T> Future<T> submit(Runnable task, T result) { };
public <T> Future<T> submit(Callable<T> task) { };
private <T> T doInvokeAny(Collection<? extends Callable<T>> tasks,
boolean timed, long nanos)
throws InterruptedException, ExecutionException, TimeoutException {
};
public <T> T invokeAny(Collection<? extends Callable<T>> tasks)
throws InterruptedException, ExecutionException {
};
public <T> T invokeAny(Collection<? extends Callable<T>> tasks,
long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException {
};
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
throws InterruptedException {
};
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
long timeout, TimeUnit unit)
throws InterruptedException {
};
}
|
AbstractExecutorService是一個抽象類，它實現了ExecutorService介面。
我們接著看ExecutorService介面的實現：
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 |
public interface ExecutorService extends Executor {
void shutdown();
boolean isShutdown();
boolean isTerminated();
boolean awaitTermination(long timeout, TimeUnit unit)
throws InterruptedException;
<T> Future<T> submit(Callable<T> task);
<T> Future<T> submit(Runnable task, T result);
Future<?> submit(Runnable task);
<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
throws InterruptedException;
<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
long timeout, TimeUnit unit)
throws InterruptedException;
<T> T invokeAny(Collection<? extends Callable<T>> tasks)
throws InterruptedException, ExecutionException;
<T> T invokeAny(Collection<? extends Callable<T>> tasks,
long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException;
}
|
而ExecutorService又是繼承了Executor介面，我們看一下Executor介面的實現：
| 1 2 3 |
public interface Executor {
void execute(Runnable command);
}
|
到這裡，大家應該明白了ThreadPoolExecutor、AbstractExecutorService、ExecutorService和Executor幾個之間的關係了。
Executor是一個頂層介面，在它裡面只聲明瞭一個方法execute(Runnable)，返回值為void，參數為Runnable類型，從字面意思可以理解，就是用來執行傳進去的任務的；
然後ExecutorService介面繼承了Executor介面，並聲明瞭一些方法：submit、invokeAll、invokeAny以及shutDown等；
抽象類AbstractExecutorService實現了ExecutorService介面，基本實現了ExecutorService中聲明的所有方法；
然後ThreadPoolExecutor繼承了類AbstractExecutorService。
在ThreadPoolExecutor類中有幾個非常重要的方法：
| 1 2 3 4 |
execute()
submit()
shutdown()
shutdownNow()
|
execute()方法實際上是Executor中聲明的方法，在ThreadPoolExecutor進行了具體的實現，這個方法是ThreadPoolExecutor的核心方法，通過這個方法可以向線程池提交一個任務，交由線程池去執行。
submit()方法是在ExecutorService中聲明的方法，在AbstractExecutorService就已經有了具體的實現，在ThreadPoolExecutor中並沒有對其進行重寫，這個方法也是用來向線程池提交任務的，但是它和execute()方法不同，它能夠返回任務執行的結果，去看submit()方法的實現，會發現它實際上還是調用的execute()方法，只不過它利用了Future來獲取任務執行結果（Future相關內容將在下一篇講述）。
shutdown()和shutdownNow()是用來關閉線程池的。
還有很多其他的方法：
比如：getQueue() 、getPoolSize() 、getActiveCount()、getCompletedTaskCount()等獲取與線程池相關屬性的方法，有興趣的朋友可以自行查閱API。
二.深入剖析線程池實現原理
在上一節我們從巨集觀上介紹了ThreadPoolExecutor，下麵我們來深入解析一下線程池的具體實現原理，將從下麵幾個方面講解：
1.線程池狀態
2.任務的執行
3.線程池中的線程初始化
4.任務緩存隊列及排隊策略
5.任務拒絕策略
6.線程池的關閉
7.線程池容量的動態調整
1.線程池狀態
在ThreadPoolExecutor中定義了一個volatile變數，另外定義了幾個static final變數表示線程池的各個狀態：
| 1 2 3 4 5 |
volatile int runState;
static final int RUNNING = 0;
static final int SHUTDOWN = 1;
static final int STOP = 2;
static final int TERMINATED = 3;
|
runState表示當前線程池的狀態，它是一個volatile變數用來保證線程之間的可見性；
下麵的幾個static final變數表示runState可能的幾個取值。
當創建線程池後，初始時，線程池處於RUNNING狀態；
如果調用了shutdown()方法，則線程池處於SHUTDOWN狀態，此時線程池不能夠接受新的任務，它會等待所有任務執行完畢；
如果調用了shutdownNow()方法，則線程池處於STOP狀態，此時線程池不能接受新的任務，並且會去嘗試終止正在執行的任務；
當線程池處於SHUTDOWN或STOP狀態，並且所有工作線程已經銷毀，任務緩存隊列已經清空或執行結束後，線程池被設置為TERMINATED狀態。
2.任務的執行
在瞭解將任務提交給線程池到任務執行完畢整個過程之前，我們先來看一下ThreadPoolExecutor類中其他的一些比較重要成員變數：
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 |
private final BlockingQueue<Runnable> workQueue; //任務緩存隊列，用來存放等待執行的任務
private final ReentrantLock mainLock = new ReentrantLock(); //線程池的主要狀態鎖，對線程池狀態（比如線程池大小
//、runState等）的改變都要使用這個鎖
private final HashSet<Worker> workers = new HashSet<Worker>(); //用來存放工作集
private volatile long keepAliveTime; //線程存貨時間
private volatile boolean allowCoreThreadTimeOut; //是否允許為核心線程設置存活時間
private volatile int corePoolSize; //核心池的大小（即線程池中的線程數目大於這個參數時，提交的任務會被放進任務緩存隊列）
private volatile int maximumPoolSize; //線程池最大能容忍的線程數
private volatile int poolSize; //線程池中當前的線程數
private volatile RejectedExecutionHandler handler; //任務拒絕策略
private volatile ThreadFactory threadFactory; //線程工廠，用來創建線程
private int largestPoolSize; //用來記錄線程池中曾經出現過的最大線程數
private long completedTaskCount; //用來記錄已經執行完畢的任務個數
|
每個變數的作用都已經標明出來了，這裡要重點解釋一下corePoolSize、maximumPoolSize、largestPoolSize三個變數。
corePoolSize在很多地方被翻譯成核心池大小，其實我的理解這個就是線程池的大小。舉個簡單的例子：
假如有一個工廠，工廠裡面有10個工人，每個工人同時只能做一件任務。
因此只要當10個工人中有工人是空閑的，來了任務就分配給空閑的工人做；
當10個工人都有任務在做時，如果還來了任務，就把任務進行排隊等待；
如果說新任務數目增長的速度遠遠大於工人做任務的速度，那麼此時工廠主管可能會想補救措施，比如重新招4個臨時工人進來；
然後就將任務也分配給這4個臨時工人做；
如果說著14個工人做任務的速度還是不夠，此時工廠主管可能就要考慮不再接收新的任務或者拋棄前面的一些任務了。
當這14個工人當中有人空閑時，而新任務增長的速度又比較緩慢，工廠主管可能就考慮辭掉4個臨時工了，只保持原來的10個工人，畢竟請額外的工人是要花錢的。
這個例子中的corePoolSize就是10，而maximumPoolSize就是14（10+4）。
也就是說corePoolSize就是線程池大小，maximumPoolSize在我看來是線程池的一種補救措施，即任務量突然過大時的一種補救措施。
不過為了方便理解，在本文後面還是將corePoolSize翻譯成核心池大小。
largestPoolSize只是一個用來起記錄作用的變數，用來記錄線程池中曾經有過的最大線程數目，跟線程池的容量沒有任何關係。
下麵我們進入正題，看一下任務從提交到最終執行完畢經歷了哪些過程。
在ThreadPoolExecutor類中，最核心的任務提交方法是execute()方法，雖然通過submit也可以提交任務，但是實際上submit方法裡面最終調用的還是execute()方法，所以我們只需要研究execute()方法的實現原理即可：
| 1 2 3 4 5 6 7 8 9 10 11 12 |
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {
if (runState == RUNNING && workQueue.offer(command)) {
if (runState != RUNNING || poolSize == 0)
ensureQueuedTaskHandled(command);
}
else if (!addIfUnderMaximumPoolSize(command))
reject(command); // is shutdown or saturated
}
}
|
上面的代碼可能看起來不是那麼容易理解，下麵我們一句一句解釋：
首先，判斷提交的任務command是否為null，若是null，則拋出空指針異常；
接著是這句，這句要好好理解一下：
| 1 |
if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command))
|
由於是或條件運算符，所以先計算前半部分的值，如果線程池中當前線程數不小於核心池大小，那麼就會直接進入下麵的if語句塊了。
如果線程池中當前線程數小於核心池大小，則接著執行後半部分，也就是執行
| 1 |
addIfUnderCorePoolSize(command)
|
如果執行完addIfUnderCorePoolSize這個方法返回false，則繼續執行下麵的if語句塊，否則整個方法就直接執行完畢了。
如果執行完addIfUnderCorePoolSize這個方法返回false，然後接著判斷：
| 1 |
if (runState == RUNNING && workQueue.offer(command))
|
如果當前線程池處於RUNNING狀態，則將任務放入任務緩存隊列；如果當前線程池不處於RUNNING狀態或者任務放入緩存隊列失敗，則執行：
| 1 |
addIfUnderMaximumPoolSize(command)
|
如果執行addIfUnderMaximumPoolSize方法失敗，則執行reject()方法進行任務拒絕處理。
回到前面：
| 1 |
if (runState == RUNNING && workQueue.offer(command))
|
這句的執行，如果說當前線程池處於RUNNING狀態且將任務放入任務緩存隊列成功，則繼續進行判斷：
| 1 |
if (runState != RUNNING || poolSize == 0)
|
這句判斷是為了防止在將此任務添加進任務緩存隊列的同時其他線程突然調用shutdown或者shutdownNow方法關閉了線程池的一種應急措施。如果是這樣就執行：
| 1 |
ensureQueuedTaskHandled(command)
|
進行應急處理，從名字可以看出是保證 添加到任務緩存隊列中的任務得到處理。
我們接著看2個關鍵方法的實現：addIfUnderCorePoolSize和addIfUnderMaximumPoolSize：
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 |
private boolean addIfUnderCorePoolSize(Runnable firstTask) {
Thread t = null;
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
if (poolSize < corePoolSize && runState == RUNNING)
t = addThread(firstTask); //創建線程去執行firstTask任務
} finally {
mainLock.unlock();
}
if (t == null)
return false;
t.start();
return true;
}
|
這個是addIfUnderCorePoolSize方法的具體實現，從名字可以看出它的意圖就是當低於核心吃大小時執行的方法。下麵看其具體實現，首先獲取到鎖，因為這地方涉及到線程池狀態的變化，先通過if語句判斷當前線程池中的線程數目是否小於核心池大小，有朋友也許會有疑問：前面在execute()方法中不是已經判斷過了嗎，只有線程池當前線程數目小於核心池大小才會執行addIfUnderCorePoolSize方法的，為何這地方還要繼續判斷？原因很簡單，前面的判斷過程中並沒有加鎖，因此可能在execute方法判斷的時候poolSize小於corePoolSize，而判斷完之後，在其他線程中又向線程池提交了任務，就可能導致poolSize不小於corePoolSize了，所以需要在這個地方繼續判斷。然後接著判斷線程池的狀態是否為RUNNING，原因也很簡單，因為有可能在其他線程中調用了shutdown或者shutdownNow方法。然後就是執行
| 1 |
t = addThread(firstTask);
|
這個方法也非常關鍵，傳進去的參數為提交的任務，返回值為Thread類型。然後接著在下麵判斷t是否為空，為空則表明創建線程失敗（即poolSize>=corePoolSize或者runState不等於RUNNING），否則調用t.start()方法啟動線程。
我們來看一下addThread方法的實現：
| 1 2 3 4 5 6 7 8 9 10 11 12 |
private Thread addThread(Runnable firstTask) {
Worker w = new Worker(firstTask);
Thread t = threadFactory.newThread(w); //創建一個線程，執行任務
if (t != null) {
w.thread = t; //將創建的線程的引用賦值為w的成員變數
workers.add(w);
int nt = ++poolSize; //當前線程數加1
if (nt > largestPoolSize)
largestPoolSize = nt;
}
return t;
}
|
在addThread方法中，首先用提交的任務創建了一個Worker對象，然後調用線程工廠threadFactory創建了一個新的線程t，然後將線程t的引用賦值給了Worker對象的成員變數thread，接著通過workers.add(w)將Worker對象添加到工作集當中。
下麵我們看一下Worker類的實現：
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 |
private final class Worker implements Runnable {
private final ReentrantLock runLock = new ReentrantLock();
private Runnable firstTask;
volatile long completedTasks;
Thread thread;
Worker(Runnable firstTask) {
this.firstTask = firstTask;
}
boolean
|
