Java線程池ThreadPoolExecuter：execute()原理

一、線程池執行任務的流程

1. 如果線程池工作線程數<corePoolSize，創建新線程執行task，並不斷輪訓t等待隊列處理task。

2. 如果線程池工作線程數>=corePoolSize並且等待隊列未滿，將task插入等待隊列。

3. 如果線程池工作流程數>=corePoolSize並且等待隊列已滿，且工作線程數<maximumPoolSize,創建新線程執行task。

4. 如果線程池工作流程數>=corePoolSize並且等待隊列已滿，且工作線程數=maximumPoolSize,執行拒絕策略。

二、execute()原理 

public void execute(Runnable command) {        if (command == null)
            throw new NullPointerException();
        /*
         * Proceed in 3 steps:
         *
         * 1. If fewer than corePoolSize threads are running, try to
         * start a new thread with the given command as its first
         * task.  The call to addWorker atomically checks runState and
         * workerCount, and so prevents false alarms that would add
         * threads when it shouldn't, by returning false.
         * 如果運行的線程數小於corePoolSize，嘗試創建一個新線程（Worker），並執行它的第一個任務command
　　　　　　
         * 2. If a task can be successfully queued, then we still need
         * to double-check whether we should have added a thread
         * (because existing ones died since last checking) or that
         * the pool shut down since entry into this method. So we
         * recheck state and if necessary roll back the enqueuing if
         * stopped, or start a new thread if there are none.
         * 如果task成功插入等待隊列，我們仍需要進行雙重校驗是否可以成功添加一個線程
　　　　　　（因為有的線程可能在我們上次檢查以后已經死掉了）或者在我們進入這個方法后線程池已經關閉了

         * 3. If we cannot queue task, then we try to add a new
         * thread.  If it fails, we know we are shut down or saturated
         * and so reject the task.
　　　　　   如果等待隊列已滿，我們嘗試新創建一個線程。如果創建失敗，我們知道線程已關閉或者已飽和，因此我們拒絕改任務。
         */
        int c = ctl.get();
　　　　  //工作線程小於核心線程數，創建新的線程
        if (workerCountOf(c) < corePoolSize) {
　　　　　　　 //創建新的worker立即執行command，並且輪訓workQueue處理task
            if (addWorker(command, true)) 
                return;
            c = ctl.get();
        }
　　　　　//線程池在運行狀態且可以將task插入隊列
　　　　　//第一次校驗線程池在運行狀態
        if (isRunning(c) && workQueue.offer(command)) {
            int recheck = ctl.get();
　　　　　　　　//第二次校驗，防止在第一次校驗通過后線程池關閉。如果線程池關閉，在隊列中刪除task並拒絕task
            if (! isRunning(recheck) && remove(command))
                reject(command);
            //如果線程數=0（線程都死掉了，比如：corePoolSize=0），新建線程且未指定firstTask，僅僅去輪訓workQueue
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
　　　　　//線程隊列已滿，嘗試創建新線程執行task，創建失敗后拒絕task
        //創建失敗原因：1.線程池關閉；2.線程數已經達到maxPoolSize
        else if (!addWorker(command, false))
            reject(command);
    }

1.  addWorker(Runnable firstTask, boolean core)

private boolean addWorker(Runnable firstTask, boolean core) {
        retry:
　　　　 //外層循環判斷線程池的狀態
        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);//線程池狀態
　　　　　
            // Check if queue empty only if necessary.
　　　　　　　//線程池狀態:RUNNING = -1、SHUTDOWN = 0、STOP = 1、TIDYING = 2、TERMINATED = 3　
            
　　　　　　　　//線程池至少是shutdown狀態
　　　　　　　 if (rs >= SHUTDOWN &&
　　　　　　　　　　//除了線程池正在關閉（shutdown），隊列里還有未處理的task的情況，其他都不能添加
                ! (rs == SHUTDOWN &&
                   firstTask == null &&
                   ! workQueue.isEmpty()))
                return false;
            //內層循環判斷是否到達容量上限，worker+1
            for (;;) {
                int wc = workerCountOf(c);//worker數量
                //worker大於Integer最大上限
                //或到達邊界上限
                if (wc >= CAPACITY ||
                    wc >= (core ? corePoolSize : maximumPoolSize))
                    return false;
                //CAS worker+1
                if (compareAndIncrementWorkerCount(c))
                    break retry;//成功了跳出循環
                c = ctl.get();  // Re-read ctl
                if (runStateOf(c) != rs) //如果線程池狀態發生變化，重試外層循環
                    continue retry;
                // else CAS failed due to workerCount change; retry inner loop 
　　　　　　　　　// CAS失敗workerCount被其他線程改變，重新嘗試內層循環CAS對workerCount+1

            }
        }

        boolean workerStarted = false;
        boolean workerAdded = false;
        Worker w = null;
        try {
            final ReentrantLock mainLock = this.mainLock;
            w = new Worker(firstTask); //1.state置為-1，Worker繼承了AbstractQueuedSynchronizer
                                       //2.設置firstTask屬性
                                       //3.Worker實現了Runable接口，將this作為入參創建線程
            final Thread t = w.thread;
            if (t != null) {
　　　　　　　　　　//addWorker需要加鎖
                mainLock.lock();
                try {
                    // Recheck while holding lock.
                    // Back out on ThreadFactory failure or if
                    // shut down before lock acquired.
                    int c = ctl.get();
                    int rs = runStateOf(c);

                    if (rs < SHUTDOWN ||
                        (rs == SHUTDOWN && firstTask == null)) {
                        if (t.isAlive()) // precheck that t is startable
                            throw new IllegalThreadStateException();
                        workers.add(w);//workers是HashSet<Worker>
　　　　　　　　　　　　　　//設置最大線程池大小
                        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;
    }

addWorker(Runnable firstTask, boolean core)
參數：
    firstTask：    worker線程的初始任務，可以為空
    core：           true：將corePoolSize作為上限，false：將maximumPoolSize作為上限

addWorker方法有4種傳參的方式：

    1、addWorker(command, true)

    2、addWorker(command, false)

    3、addWorker(null, false)

    4、addWorker(null, true)

在execute方法中就使用了前3種，結合這個核心方法進行以下分析

1、線程數小於corePoolSize。判斷workers(HashSet<Worker>)大小，如果worker數量>=corePoolSize 返回false，否則創建worker添加到workers，並執行worker的run方法（執行firstTask並輪詢tworkQueue）；

2、線程數大於corePoolSize且workQueue已滿。如果worker數量>=maximumPoolSize返回false，否則創建worker添加到workers，並執行worker的run方法（執行firstTask並輪詢tworkQueue）；

3.、沒有worker存活，創建worker去輪詢workQueue，長度限制maximumPoolSize。

4、prestartAllCoreThreads()方法調用，啟動所有的核心線程去輪詢workQueue。因為addWorker是需要上鎖的，預啟動核心線程可以提高執行效率。

2.　ThreadPoolExecutor 內部類Worker 

 /** 
　　　* Class Worker mainly maintains interrupt control state for
 * threads running tasks, along with other minor bookkeeping. * This class opportunistically extends AbstractQueuedSynchronizer * to simplify acquiring and releasing a lock surrounding each * task execution. This protects against interrupts that are * intended to wake up a worker thread waiting for a task from * instead interrupting a task being run. We implement a simple * non-reentrant mutual exclusion lock rather than use * ReentrantLock because we do not want worker tasks to be able to * reacquire the lock when they invoke pool control methods like * setCorePoolSize. Additionally, to suppress interrupts until * the thread actually starts running tasks, we initialize lock * state to a negative value, and clear it upon start (in * runWorker).
　　　* 1.Worker類主要負責運行線程狀態的控制。
　　　* 2.Worker繼承了AQS實現了簡單的獲取鎖和釋放所的操作。來避免中斷等待執行任務的線程時，中斷正在運行中的線程（線程剛啟動，還沒開始執行任務）。
　　　* 3.自己實現不可重入鎖，是為了避免在實現線程池控狀態控制的方法，例如：setCorePoolSize的時候中斷正在開始運行的線程。
　　 * setCorePoolSize可能會調用interruptIdleWorkers()，該方法中會調用worker的tryLock()方法中斷線程，自己實現鎖可以確保工作線程啟動之前不會被中斷 */ private final class Worker extends AbstractQueuedSynchronizer implements Runnable { /** * This class will never be serialized, but we provide a * serialVersionUID to suppress a javac warning. */ private static final long serialVersionUID = 6138294804551838833L; /** Thread this worker is running in. Null if factory fails. */ final Thread thread; /** Initial task to run. Possibly null. */ Runnable firstTask; /** Per-thread task counter */ volatile long completedTasks; /** * Creates with given first task and thread from ThreadFactory. * @param firstTask the first task (null if none) */ Worker(Runnable firstTask) { setState(-1); // inhibit interrupts until runWorker //狀態置為-1，如果中斷線程需要CAS將state 從0->1，以此來保證能只中斷從workerQueue getTask的線程 this.firstTask = firstTask; this.thread = getThreadFactory().newThread(this); } /** Delegates main run loop to outer runWorker */ public void run() { runWorker(this); //首先執行w.unlock,就是把state置為0，對該線程的中斷就可以進行了 } // Lock methods // // The value 0 represents the unlocked state. // The value 1 represents the locked state. protected boolean isHeldExclusively() { return getState() != 0; } //在setCorePoolSize/shutdown等方法中斷worker線程時需要調用該方法，確保中斷的是從workerQueue getTask的線程 protected boolean tryAcquire(int unused) { if (compareAndSetState(0, 1)) { setExclusiveOwnerThread(Thread.currentThread()); return true; } return false; } protected boolean tryRelease(int unused) { setExclusiveOwnerThread(null); setState(0); return true; } public void lock() { acquire(1); } public boolean tryLock() { return tryAcquire(1); } public void unlock() { release(1); } //調用tryRelease修改state=0，LockSupport.unpark(thread)下一個等待鎖的線程 public boolean isLocked() { return isHeldExclusively(); } void interruptIfStarted() { Thread t; if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) { try { t.interrupt(); } catch (SecurityException ignore) { } } } }