java線程池源碼淺析

  最近工作不算太忙，抽時間學習了下java線程池底層源碼，廢話不多說，馬上“去片”！

  Executors類是java線程池的工具類，此類位於java.util.concurrent包下。在日常項目開發中，我們使用得比較多的主要有CachedThreadPool、FixedThreadPool、SingleThreadExecutor、ScheduledThreadPool這4個線程池，這些線程池底層均調用new ThreadPoolExecutor()方法來新建一個線程池。

現在簡單介紹這4個池：

CachedThreadPool

CachedThreadPool默認最大線程數量為int的最大值，狹義來說已經可以說是不限制數量了。默認的空閑線程終止時間60秒。一般在項目開發中如果不明確線程池中運行任務的數量不建議使用此線程池，因為如果大量的並發任務運行時，會大量占用CPU資源，導致應用程序崩潰甚至服務器崩潰等嚴重問題。

FixedThreadPool

FixedThreadPool需要指定線程數量，工作任務超過線程池總線程數后會進入等待狀態 ，直到已經在工作的線程完成任務后釋放后再進行另一任務的工作。此線程池本人在日常開發中使用得較多，可以有效控制線程的數量，防止大量占用CPU資源，導致應用程序崩潰甚至服務器崩潰等問題的發生。

SingleThreadExecutor

SingleThreadExecutor根據命名就可以猜這是一個單線程線程池，如果任務超過一個時，則會排隊等待線程池唯一的線程完成任務再運行其它任務。

ScheduledThreadPool

ScheduledThreadPool為一個延時任務線程池。（在此不作詳細介紹，此線程池將作為另一篇文章單獨講解）

以上簡單介紹了java常用的4個線程池，下面將深入分析線程池創建后，是如何執行任務及線程池的底層是如何實現的。

剛剛也提到了所有線程池都要使用new ThreadPoolExecutor()方法來新建，讓我們先看一下JDK的源碼。

    public ThreadPoolExecutor(int corePoolSize,  2                               int maximumPoolSize,  3                               long keepAliveTime,  4  TimeUnit unit,  5                               BlockingQueue<Runnable> workQueue,  6  ThreadFactory threadFactory,  7  RejectedExecutionHandler handler) {  8         if (corePoolSize < 0 ||
            maximumPoolSize <= 0 ||
            maximumPoolSize < corePoolSize ||
            keepAliveTime < 0) 12             throw new IllegalArgumentException(); 13         if (workQueue == null || threadFactory == null || handler == null) 14             throw new NullPointerException(); 15         this.corePoolSize = corePoolSize; 16         this.maximumPoolSize = maximumPoolSize; 17         this.workQueue = workQueue; 18         this.keepAliveTime = unit.toNanos(keepAliveTime); 19         this.threadFactory = threadFactory; 20         this.handler = handler; 21     }

 

先看一下需要傳遞的參數：

corePoolSize 線程池核心線程數，如果當前運行線程數量小於核心線程數量的話，處理新任務的時候，線程池還是會新建一個線程執行任務，直到當前運行線程數量等於核心線程數量；

maximumPoolSize： 線程池最大線程數，當運行線程數量等於線程池最大線程數時，再來新的任務會觸發異常；

keepAliveTime： 空閑線程終止時間，線程執行完任務空閑時間超過指定時間后，線程將會被終止;

Unit： 空閑線程終止時間單位；

workQueue： 線程工作任務隊列，線程主要使用ArrayBlockingQueue，DelayQueue，DelayedWorkQueue，LinkedBlockingDeque，PriorityBlockingQueue，SynchronousQueue，LinkedTransferQueue這些隊列，這里不作詳細介紹，有興趣的童鞋可以自行谷歌學習；

threadFactory： 線程工廠；

Handler： 線程任務拒絕策略處理器，默認使用AbortPolicy策略，當任務超過最大線程時，會拋出異常。除此之外還有另外幾種異常處理策略，CallerRunsPolicy使用調用者本身的線程來執行任務，DiscardPolicy丟棄當前任務，DiscardOldestPolicy丟棄最舊的任務，執行當前任務；

 

介紹完新建線程池的基本參數傳遞，現在到了本文最核心部分，線程池調用 execute方法時，內部究竟是如何實現的。

我們先看execute方法的源碼：

public void execute(Runnable command) {
//先判斷任務是否為空,如果為空則拋出異常；    

if (command == null)
        throw new NullPointerException();

//獲取當前運行線程數，與線程池核心線程數作對比，如果當前運行線程數小於核心線程數，則調用addWorker方法將任務加入隊列執行任務。這是一個任務加入線程池運行最基本的過程。
    int c = ctl.get();
    if (workerCountOf(c) < corePoolSize) {
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    if (isRunning(c) && workQueue.offer(command)) {
        int recheck = ctl.get();
        if (! isRunning(recheck) && remove(command))
            reject(command);
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    else if (!addWorker(command, false))
        reject(command);
}

 

后面一段源碼我們待會再分析，先看addWorker方法做了什么，以下為源碼：

private boolean addWorker(Runnable firstTask, boolean core) {
    //首先看到這是一個無限循環
    retry:
    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);

        //檢查任務和隊列是否為空，如果為空則返回false
        if (rs >= SHUTDOWN &&
            ! (rs == SHUTDOWN &&
               firstTask == null &&
               ! workQueue.isEmpty()))
            return false;

        for (;;) {

//此處檢查當前運行的任務數，參數core在此起作用，如果true則比較當前任務數是否大於等於線程池核心線程數，false則比較當前任務數是否大於等於線程池總線程數。
            int wc = workerCountOf(c);
            if (wc >= CAPACITY ||
                wc >= (core ? corePoolSize : maximumPoolSize))
                return false;

//此處使用AtomicInteger類compareAndSet方法（此方法為原子性）將當前運行線程數加1，成功加1則退出循環
            if (compareAndIncrementWorkerCount(c))
                break retry;

//由於此循環代碼段沒有加鎖，為防止多個線程並發引起數據問題，這里再次檢查當前運行任務數
            c = ctl.get();  
            if (runStateOf(c) != rs)
                continue retry;
                    }
    }

    boolean workerStarted = false;
    boolean workerAdded = false;
    Worker w = null;
    try {
        final ReentrantLock mainLock = this.mainLock;
        //此處新建一個Worker,Worker繼承了Runnable接口，用作執行實際任務用

w = new Worker(firstTask);
        final Thread t = w.thread;
        if (t != null) {

//此處加鎖，防止並發
            mainLock.lock();
            try {
                int c = ctl.get();
                int rs = runStateOf(c);
     //檢查任務是否為空
                if (rs < SHUTDOWN ||
                    (rs == SHUTDOWN && firstTask == null)) {

//檢查線程是否已經啟動，如果是則拋出異常
                    if (t.isAlive())

                        throw new IllegalThreadStateException();
                    workers.add(w);
                    int s = workers.size();
                    if (s > largestPoolSize)
                        largestPoolSize = s;
                    workerAdded = true;
                }
            } finally {
                mainLock.unlock();
            }

  //最后啟動線程，執行任務
            if (workerAdded) {
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        if (! workerStarted)
            addWorkerFailed(w);
    }
    return workerStarted;
}

 

以上就是線程池新建一個線程，成功執行一個任務的過程。那當線程池中的線程已經達到最大數量了，線程池又會是如何工作，並觸發任務拒絕策略的呢？請看以下源碼：

//當前工作任務數如果大於等於核心線程數的話，將會執行以下代碼

if (isRunning(c) && workQueue.offer(command)) {
        int recheck = ctl.get();

//如果任務不是正在運行並且能將任務從隊列移除，執行reject方法，根據任務拒絕策略進行處理
        if (! isRunning(recheck) && remove(command))
            reject(command);

//如果當前運行的任務為0，傳入一個空的任務
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }

//如果添加任務到隊列失敗,執行reject方法，根據任務拒絕策略進行處理
    else if (!addWorker(command, false))

     reject(command);
}

 

  以上就是我最近在學習線程池及分析其源碼得到的“成果”，如有錯漏這處，歡迎指正！

（如轉載請附上原文地址：https://www.cnblogs.com/kenblog/p/9366691.html ）