執行緒池的原始碼及原理[JDK1.6實現]

1.執行緒池的包含的內容

2.執行緒池的資料結構【核心類ThreadPoolExecutor】：

worker：工作類，一個worker代表啟動了一個執行緒，它啟動後會迴圈執行workQueue裡面的所有任務 workQueue：任務佇列，用於存放待執行的任務 keepAliveTime：執行緒活動保持時間，執行緒池的工作執行緒空閒後，保持存活的時間。 執行緒池原理：預先啟動一些執行緒，執行緒無限迴圈從任務佇列中獲取一個任務進行執行，直到執行緒池被關閉。如果某個執行緒因為執行某個任務發生異常而終止，那麼重新建立一個新的執行緒而已。如此反覆。

3.執行緒池任務submit及執行流程

a.一個任務提交，如果執行緒池大小沒達到corePoolSize，則每次都啟動一個worker也就是一個執行緒來立即執行 b.如果來不及執行，則把多餘的執行緒放到workQueue，等待已啟動的worker來迴圈執行 c.如果佇列workQueue都放滿了還沒有執行，則在maximumPoolSize下面啟動新的worker來迴圈執行workQueue d.如果啟動到maximumPoolSize還有任務進來，執行緒池已達到滿負載，此時就執行任務拒絕RejectedExecutionHandler  Java Code 執行緒池核心的程式碼

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24

// 流程就是：沒達到corePoolSize，建立worker執行，達到corePoolSize加入workQueue //           workQueue滿了且在maximumPoolSize下，建立新worker，達到maximumPoolSize，執行reject public void execute(Runnable command) {     if (command == null)         throw new NullPointerException();     // 1：poolSize達到corePoolSize，執行3把任務加入workQueue     // 2：poolSize沒達到，執行addIfUnderCorePoolSize()在corePoolSize內建立新worker立即執行任務     //    如果達到corePoolSize，則同上執行3     if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {         // 3:workQueue滿了，執行5         if (runState == RUNNING && workQueue.offer(command)) {             if (runState != RUNNING || poolSize == 0) {                 // 4:如果執行緒池關閉，執行拒絕策略                 //   如果poolSize==0，新啟動一個執行緒執行佇列內任務                 ensureQueuedTaskHandled(command);             }             // 5:在maximumPoolSize內建立新worker立即執行任務             //   如果達到maximumPoolSize，執行6拒絕策略         } else if (!addIfUnderMaximumPoolSize(command))             // 6:拒絕策略             reject(command); // is shutdown or saturated     } }

從上面程式碼可以看出，一個任務提交什麼時候立即執行呢？ runState == RUNNING && ( poolSize < corePoolSize || (workQueue.isFull() && poolSize < maxnumPoolSize))

4.工作Worker的原理

上面講過執行緒池建立執行緒其實是委託給Worker這個物件完成的。worker會迴圈獲取工作佇列的任務來完成  Java Code 工作執行緒Worker執行

1 2 3 4 5 6 7 8 9 10 11 12 13 14

public void run() {     try {         Runnable task = firstTask;         firstTask = null;         // getTask()是從workQueue裡面阻塞獲取任務，如果getTask()返回null則終結本執行緒         while (task != null || (task = getTask()) != null) {             runTask(task);             task = null;         }     } finally {         // 走到這裡代表這個worker或者說這個執行緒由於執行緒池關閉或超過aliveTime需要關閉了         workerDone(this);     } }

5.執行緒的銷燬

keepAliveTime：代表的就是執行緒空閒後多久後銷燬，執行緒的銷燬是通過worker的getTask()來實現的。 一般來說，Worker會迴圈獲取getTask()，如果getTask()返回null則工作執行緒worker終結，那我們再看看什麼時候getTask()返回null  Java Code Worker的getTask方法

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

Runnable getTask() {     for (;;) {         try {             int state = runState;             if (state > SHUTDOWN)                 return null;             Runnable r;             if (state == SHUTDOWN)  // Help drain queue                 r = workQueue.poll();             else if (poolSize > corePoolSize || allowCoreThreadTimeOut)                 // 在poolSize大於corePoolSize或允許核心執行緒超時時                 // 阻塞超時獲取有可能獲取到null，此時worker執行緒銷燬                 r = workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS);             else                 r = workQueue.take();             if (r != null)                 return r;             // 這裡是是否執行worker執行緒銷燬的判斷             if (workerCanExit()) {                 if (runState >= SHUTDOWN)                     // STOP或TERMINATED狀態，終止空閒worker                     interruptIdleWorkers();                 return null; // 這裡返回null，代表工作執行緒worker銷燬             }             // 其他：retry，繼續迴圈         } catch (InterruptedException ie) {             // On interruption, re-check runState         }     } }

6.執行緒池關閉

平緩關閉 shutdown：已經啟動的任務全部執行完畢，同時不再接受新的任務 立即關閉 shutdownNow：取消所有正在執行和未執行的任務 具體參考原始碼

7.執行緒池的監控

通過執行緒池提供的引數進行監控。執行緒池裡有一些屬性在監控執行緒池的時候可以使用

taskCount：執行緒池需要執行的任務數量。 completedTaskCount：執行緒池在執行過程中已完成的任務數量。小於或等於taskCount。 largestPoolSize：執行緒池曾經建立過的最大執行緒數量。通過這個資料可以知道執行緒池是否滿過。如等於執行緒池的最大大小，則表示執行緒池曾經滿了。 getPoolSize:執行緒池的執行緒數量。如果執行緒池不銷燬的話，池裡的執行緒不會自動銷燬，所以這個大小隻增不+ getActiveCount：獲取活動的執行緒數。

通過擴充套件執行緒池進行監控。通過繼承執行緒池並重寫執行緒池的beforeExecute，afterExecute和terminated方法，我們可以在任務執行前，執行後和執行緒池關閉前幹一些事情。如監控任務的平均執行時間，最大執行時間和最小執行時間等。這幾個方法線上程池裡是空方法。

8.執行緒池調優[更多可參考：執行緒池與工作佇列]

調整執行緒池的大小 - 執行緒池的最佳大小取決於可用處理器的數目以及工作佇列中的任務的性質。 調整執行緒池的大小基本上就是避免兩類錯誤：執行緒太少或執行緒太多。

a.CPU限制的任務，提高CPU利用率。

在運行於具有 N 個處理器機器上的計算限制的應用程式中，線上程數目接近 N 時新增額外的執行緒可能會改善總處理能力，而線上程數目超過 N 時新增額外的執行緒將不起作用。事實上，太多的執行緒甚至會降低效能，因為它會導致額外的環境切換開銷。      若在一個具有 N 個處理器的系統上只有一個工作佇列，其中全部是計算性質的任務，線上程池具有N 或 N+1個執行緒時一般會獲得最大的 CPU 利用率。

b.I/O限制的任務（例如，從套接字讀取 HTTP 請求的任務）

需要讓池的大小超過可用處理器的數目，因為並不是所有執行緒都一直在工作。通過使用概要分析，您可以或得一些資料，並計算出大概的執行緒池大小。 Amdahl 法則提供很好的近似公式。用 WT 表示每項任務的平均等待時間，ST 表示每項任務的平均服務時間（計算時間）。則 WT/ST 是每項任務等待所用時間的百分比。對於 N 處理器系統，池中可以近似有N*(1+WT/ST) 個執行緒。 

c.綜合考慮執行緒池效能瓶頸

a.處理器利用率 b.隨著執行緒池的增長，您可能會碰到排程程式、可用記憶體方面的限制，或者其它系統資源方面的限制，例如套接字、開啟的檔案控制代碼或資料庫連線等的數目。

9.執行緒池擴充套件 - 延時執行緒池 ScheduledThreadPoolExecutor

ScheduledThreadPoolExecutor是在普通執行緒池的基礎上增加了兩個功能，一是延時執行+定時執行，二是重複執行 定時Executor的流程在大體上與普通執行緒池一致，因此它繼承於ThreadPoolExecutor，對於問題1，它採用了DelayedQueue來實現此功能。對於問題2，定時Executor每次執行完呼叫ThreadPoolExecutor.runAndReset()重置狀態，然後重新把任務加入到Delayed佇列中     定時Executor在外部Runnable的基礎上套了一個ScheduledFutureTask，其核心原始碼如下：  Java Code 普通任務的外部封裝Future

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

// 加入的任務外部封裝了ScheduledFutureTask，繼承於FutureTask，因此也可以獲取任務結果 private class ScheduledFutureTask<V> extends FutureTask<V> implements RunnableScheduledFuture<V> {     // 省略部分程式碼     // 週期性執行，執行完成就把任務加入到delay佇列中     private void runPeriodic() {         // 這裡重置執行緒池狀態         boolean ok = ScheduledFutureTask.super.runAndReset();         boolean down = isShutdown();         // Reschedule if not cancelled and not shutdown or policy allows         if (ok && (!down ||                    (getContinueExistingPeriodicTasksAfterShutdownPolicy() &&                     !isTerminating()))) {             long p = period;             if (p > 0)                 time += p;             else                 time = now() - p;             // 重複把任務加入到執行緒池delay佇列中             ScheduledThreadPoolExecutor.super.getQueue().add(this);         } else if (down)             interruptIdleWorkers();     }     // 執行緒池呼叫的run方法     public void run() {         if (isPeriodic())             runPeriodic();         else             ScheduledFutureTask.super.run();     } }

10.執行緒池擴充套件 - ExecutorCompletionService

ExecutorCompletionService描述的是：我們提交一堆任務到執行緒池，執行緒池在任務執行完成後把結果FutureTask放到FIFO佇列中，然後我們就可以來對結果進行操作，比如說彙總，累加等等。 其實現原理很簡單，就是利用FutureTask的done()擴充套件方法，把此Future加入到queue，請看原始碼  Java Code ExecutorCompletionService核心程式碼

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

public class ExecutorCompletionService<V> implements CompletionService<V> {     // 部分程式碼省略     // 外部Future的封裝類     private class QueueingFuture extends FutureTask<Void> {         QueueingFuture(RunnableFuture<V> task) {             super(task, null);             this.task = task;         }         // 這裡把Future加入到 completionQueue         protected void done() {             completionQueue.add(task);         }         private final Future<V> task;     }     public Future<V> submit(Callable<V> task) {         if (task == null) throw new NullPointerException();         RunnableFuture<V> f = newTaskFor(task);         // 對f外層又包了一層 QueueingFuture         executor.execute(new QueueingFuture(f));         return f;     }     // 外部則可通過 completionQueue 來獲取已完成的任務Future     public Future<V> take() throws InterruptedException {         return completionQueue.take();     } }