1. 程式人生 > >ThreadPoolExecutor原始碼解析(二)

ThreadPoolExecutor原始碼解析(二)

1.ThreadPoolExcuter執行例項

首先我們先看如何新建一個ThreadPoolExecutor去執行執行緒。然後深入到原始碼中去看ThreadPoolExecutor裡面使如何運作的。

public class Test {
    public static void main(String[] args){
        /**
         * 新建一個執行緒池
         * corePoolSize:2
         * maximumPoolSize:10
         * keepAliveTime:20
         * unit:TimeUnit.SECONDS(秒)
         * workQueue:new ArrayBlockingQueue(10)
         * threadFactory:預設
         * RejectedExecutionHandler預設
         */
        ThreadPoolExecutor threadPool = new ThreadPoolExecutor(2,10,20, TimeUnit.SECONDS,new ArrayBlockingQueue(10));
        //用execute新增一個執行緒
        threadPool.execute(new Runnable() {
            @Override
            public void run() {
                try {
                    Thread.sleep(5000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
    }
}

2.ThreadPoolExecute.execute方法

可以發現,其實使用執行緒池就是使用這個方法,然後我們看這個方法具體的程式碼。


    /**
     * 在後面執行給定任務。任務在一個新的執行緒中或一個存在的worker的執行緒池中執行。
     * 如果一個執行緒不能提交到excution,可能是因為這個excutor已經shundown或者因為其容量已經是最大,
     * 此時任務將會被RejectedExecutionHandler處理
     *
     */
    public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        /*
         * Proceed in 3 steps:
         * 有以下3個步驟
         *
         * 1.如果少於corePoolSize的執行緒在執行,那麼試著啟動一個新執行緒,其中用給定指令作為first task。
         * 這會呼叫addWorker去原子性得檢查runState和workerCoune,因此可以防止錯誤報警,在錯誤報警不應該時通過返回false來新增執行緒
         * 2.如果任務被成功排隊,我們任然應該第二次檢查是否新增一個新執行緒(因為可能存在在最後一次檢查後掛掉的情況)
         * 或者在進入這個方法期間執行緒池shutdown。所以我們再次檢查狀態,如果已關閉和有必要則退出佇列,或者如果沒有的話就開始一個新的執行緒。
         * 3.如果我們無法將task入隊,那麼我們試圖新增新執行緒。如果失敗,那麼知道我們shutdown或者是飽和的並拒絕task。
         */
        int c = ctl.get();
        //判斷是否小於corePoolSize
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
        //如果pool在執行並且能提交到佇列
        if (isRunning(c) && workQueue.offer(command)) {
            int recheck = ctl.get();
            //這裡進行再次檢查,如果執行緒池沒在執行並且成功刪除task後,使用拒絕策略拒絕該task
            if (! isRunning(recheck) && remove(command))
                reject(command);
       //如果已經將task新增到佇列中,而此時沒有worker的話,那麼新建一個worker。稍後這個空閒的worker就會自動去佇列裡面取任務來執行
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
        //如果無法提交那麼按照拒絕策略拒絕task
        else if (!addWorker(command, false))
            reject(command);
    }

執行緒池的基礎和Worker基本介紹在前一節已經有說過,可以點這裡檢視。可以看到這個方法的主要流程,其實都在註釋裡面說明了。可以發現裡面主要呼叫了一個方法,addWorker()。 那麼這個addWorker()又是什麼東西呢。其實看方法名就很清楚了,就是新建一個Worker來執行你新增進來的task。

3.ThreadPoolExecute.addWorker()方法

/**
     * 檢查當前的執行緒池狀態和容量,是否可以讓一個新的worker加入。如果可以,worker計數將會被調整,並且
     * 如果可能,一個新的woker將會被建立和開始,將它當作第一個任務來執行。當執行緒池是stopped或shutdown狀態時,
     * 將返回false。當執行緒工廠建立失敗而返回null或者丟擲exception(比如典型的OOM)時,它也會返回fails。
     * firstTask:新執行緒應該第一個執行的任務。當執行緒數少於corePoolSize時或是佇列滿時,workers使用一個初始化的
     * first task來建立,用來進行分流。初始化空閒執行緒通常使用prestartCoreThread。
     * core:為true,如果使用有界的corePoolSize,否則時maxPoolSize
     * @return true if successful
     * 新增Worker
     */
    private boolean addWorker(Runnable firstTask, boolean core) {
        retry:
        for (;;) {
            int c = ctl.get();
            //狀態值
            int rs = runStateOf(c);

            // Check if queue empty only if necessary.
            //關於狀態值的檢測
            if (rs >= SHUTDOWN &&
                    ! (rs == SHUTDOWN &&
                            firstTask == null &&
                            ! workQueue.isEmpty()))
                return false;

            for (;;) {
                int wc = workerCountOf(c);
                //關於容量的檢測
                if (wc >= CAPACITY ||
                        wc >= (core ? corePoolSize : maximumPoolSize))
                    return false;
                //用到了原子CAS方法比較,使用CAS增加worker計數器成功,才能進入下一步
                if (compareAndIncrementWorkerCount(c))
                    break retry;
                //重新獲取ctl
                c = ctl.get();  // Re-read ctl
                //這裡表示執行到這裡的時候執行緒池的執行狀態改變,需要重新跳到retry處執行
                if (runStateOf(c) != rs)
                    continue retry;
                // else CAS failed due to workerCount change; retry inner loop
            }
        }

        boolean workerStarted = false;
        boolean workerAdded = false;
        Worker w = null;
        try {
            //使用firstTask初始化Worker,first可能為null,那麼則表示該worker為空閒
            w = new Worker(firstTask);
            final Thread t = w.thread;
            if (t != null) {
                final ReentrantLock mainLock = this.mainLock;
                mainLock.lock();
                try {
                    // Recheck while holding lock.
                    // Back out on ThreadFactory failure or if
                    // shut down before lock acquired.
                    //持有鎖之後再次檢查,確保一致性
                    int rs = runStateOf(ctl.get());

                    if (rs < SHUTDOWN ||
                            (rs == SHUTDOWN && firstTask == null)) {
                        if (t.isAlive()) // precheck that t is startable
                            throw new IllegalThreadStateException();
                        workers.add(w);
                        int s = workers.size();
                        //largestPoolSize為跟蹤的目前最大執行緒數,因為之前已經做過判斷,所以不會越界問題
                        if (s > largestPoolSize)
                            largestPoolSize = s;
                        workerAdded = true;
                    }
                } finally {
                    mainLock.unlock();
                }
                //workerAdded是在上面最後才設定的,確保這個變數能準確表示是否新增worker成功
                if (workerAdded) {
                    t.start();
                    workerStarted = true;
                }
            }
        } finally {
            //再次檢查
            if (! workerStarted)
                addWorkerFailed(w);
        }
        return workerStarted;
    }

addWorker本事只是為執行緒池新增一個Worker,其本身所做的事情其實很簡單,但難就難在要確保安全有效得新增一個Worker。為此addWorker()方法做了很多額外的工作。比如判斷執行緒池的執行狀態,當前Worker數量是否已經飽和等等。可以發現在這個方法,或者說整個ThreadPoolExecutor中,很多時候都是使用雙重檢查的方式來對執行緒池狀態進行檢查。其實這都是為了效率,最簡單不過直接使用Synchronized或ReentranLock進行同步,但這樣效率會低很多,所以在這裡,只有在萬不得已的情況下,才會使用悲觀的ReentranLock。

addWorker的最後直接呼叫了t.start,這裡的t其實就是Worker它自己。接下來再看Worker是如何執行的。

4.ThreadPoolExecute.runWorker()方法

/**
     * 主要的Worker執行的迴圈。重複得獲取從任務佇列中取出task並執行它。
    */

    final void runWorker(Worker w) {
        Thread wt = Thread.currentThread();
        //取出firstTask,再將worker中的值-設定為null
        Runnable task = w.firstTask;
        w.firstTask = null;
        w.unlock(); // allow interrupts
        boolean completedAbruptly = true;
        try {
            //不斷迴圈取出執行緒執行
            while (task != null || (task = getTask()) != null) {
                w.lock();
                //鎖住執行緒
                // If pool is stopping, ensure thread is interrupted;
                // if not, ensure thread is not interrupted.  This
                // requires a recheck in second case to deal with
                // shutdownNow race while clearing interrupt
                //如果當前執行緒是stop,那麼將確認其為interrupted
                if ((runStateAtLeast(ctl.get(), STOP) ||
                        (Thread.interrupted() &&
                                runStateAtLeast(ctl.get(), STOP))) &&
                        !wt.isInterrupted())
                    wt.interrupt();
                try {
                    //呼叫鉤子
                    beforeExecute(wt, task);
                    Throwable thrown = null;
                    try {
                        //執行
                        task.run();
                    } catch (RuntimeException x) {
                        thrown = x; throw x;
                    } catch (Error x) {
                        thrown = x; throw x;
                    } catch (Throwable x) {
                        thrown = x; throw new Error(x);
                    } finally {
                        afterExecute(task, thrown);
                    }
                } finally {
                    task = null;
                    w.completedTasks++;
                    w.unlock();
                }
            }
            completedAbruptly = false;
        } finally {
            processWorkerExit(w, completedAbruptly);
        }
    }

從原始碼中可以看出,一個Worker的工作其實就是不斷使用getTask()方法從佇列中獲取新的任務來執行。值得一提的是,初始化引數裡面的時間戳引數就是在這個方法裡面運用的。在迴圈體中每次都使用鎖以保證當前worker在執行task過程中不會被中斷。同時執行時還會去呼叫兩個內建的鉤子:beforeExecute()和afterExecute(),這兩個方法預設實現時空的。

同時在執行的迴圈中每次都關注著ThreadPoolExecutor的執行狀態,當執行緒池處於中斷狀態時,迴圈Worker的當前執行緒也會中斷。

總結:說到這裡就差不多把執行緒池執行task的流程說完了,當然其中忽略了很多的細節。但總而言之,ThreadPoolExecutor其實就是對worker進行管理,然後使用這些worker來執行使用者提交的task。對使用者提交的task的數量也進行一定的控制管理,比如超過一定數量時放入一個任務佇列中等等。然後對執行緒池規定一些狀態量,根據這些狀態量對執行緒池進行控制。

如果覺得對你有幫助,不如花0.5元請作者吃顆糖,讓他甜一下吧~~
![](https://img2018.cnblogs.com/blog/1011838/201809/1011838-20180913204127142-1968807170.png)