深刻淺出讓你理解Java線程池—ThreadPoolExecutor

幾句閒扯:首先,我想說java的線程池真的是很繞,之前一直都感受新建幾個線程一直不退出究竟是怎麼實現的,也就有了後來學習ThreadPoolExecutor源碼。學習源碼的過程當中,最噁心的其實就是幾種狀態的轉換了,這也是ThreadPoolExecutor的核心。花了將近小一週才大體的弄明白ThreadPoolExecutor的機制,遂記錄下來。java

線程池有多重要

線程是一個程序員必定會涉及到的一個概念,可是線程的建立和切換都是代價比較大的。因此,咱們有沒有一個好的方案能作到線程的複用呢?這就涉及到一個概念——線程池。合理的使用線程池可以帶來3個很明顯的好處:程序員

  1. 下降資源消耗:經過重用已經建立的線程來下降線程建立和銷燬的消耗
  2. 提升響應速度:任務到達時不須要等待線程建立就能夠當即執行。
  3. 提升線程的可管理性:線程池能夠統一管理、分配、調優和監控。

java多線程池的支持——ThreadPoolExecutor

java的線程池支持主要經過ThreadPoolExecutor來實現,咱們使用的ExecutorService的各類線程池策略都是基於ThreadPoolExecutor實現的,因此ThreadPoolExecutor十分重要。要弄明白各類線程池策略,必須先弄明白ThreadPoolExecutor。安全

一、實現原理

首先看一個線程池的流程圖:bash

  • step1.調用ThreadPoolExecutor的execute提交線程,首先檢查CorePool,若是CorePool內的線程小於CorePoolSize,新建立線程執行任務。
  • step2.若是當前CorePool內的線程大於等於CorePoolSize,那麼將線程加入到BlockingQueue。
  • step3.若是不能加入BlockingQueue,在小於MaxPoolSize的狀況下建立線程執行任務。
  • step4.若是線程數大於等於MaxPoolSize,那麼執行拒絕策略。

二、線程池的建立

線程池的建立能夠經過ThreadPoolExecutor的構造方法實現:markdown

/**
     * Creates a new {@code ThreadPoolExecutor} with the given initial
     * parameters.
     *
     * @param corePoolSize the number of threads to keep in the pool, even
     *        if they are idle, unless {@code allowCoreThreadTimeOut} is set
     * @param maximumPoolSize the maximum number of threads to allow in the
     *        pool
     * @param keepAliveTime when the number of threads is greater than
     *        the core, this is the maximum time that excess idle threads
     *        will wait for new tasks before terminating.
     * @param unit the time unit for the {@code keepAliveTime} argument
     * @param workQueue the queue to use for holding tasks before they are
     *        executed.  This queue will hold only the {@code Runnable}
     *        tasks submitted by the {@code execute} method.
     * @param threadFactory the factory to use when the executor
     *        creates a new thread
     * @param handler the handler to use when execution is blocked
     *        because the thread bounds and queue capacities are reached
     * @throws IllegalArgumentException if one of the following holds:<br>
     *         {@code corePoolSize < 0}<br>
     *         {@code keepAliveTime < 0}<br>
     *         {@code maximumPoolSize <= 0}<br>
     *         {@code maximumPoolSize < corePoolSize}
     * @throws NullPointerException if {@code workQueue}
     *         or {@code threadFactory} or {@code handler} is null
     */
    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler) {
        if (corePoolSize < 0 ||
            maximumPoolSize <= 0 ||
            maximumPoolSize < corePoolSize ||
            keepAliveTime < 0)
            throw new IllegalArgumentException();
        if (workQueue == null || threadFactory == null || handler == null)
            throw new NullPointerException();
        this.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.workQueue = workQueue;
        this.keepAliveTime = unit.toNanos(keepAliveTime);
        this.threadFactory = threadFactory;
        this.handler = handler;
    }
複製代碼

具體解釋一下上述參數:多線程

  1. corePoolSize 核心線程池大小
  2. maximumPoolSize 線程池最大容量大小
  3. keepAliveTime 線程池空閒時,線程存活的時間
  4. TimeUnit 時間單位
  5. ThreadFactory 線程工廠
  6. BlockingQueue任務隊列
  7. RejectedExecutionHandler 線程拒絕策略

三、線程的提交

ThreadPoolExecutor的構造方法如上所示,可是隻是作一些參數的初始化,ThreadPoolExecutor被初始化好以後即可以提交線程任務,線程的提交方法主要是execute和submit。這裏主要說execute,submit會在後續的博文中分析。less

/**
     * Executes the given task sometime in the future.  The task
     * may execute in a new thread or in an existing pooled thread.
     *
     * If the task cannot be submitted for execution, either because this
     * executor has been shutdown or because its capacity has been reached,
     * the task is handled by the current {@code RejectedExecutionHandler}.
     *
     * @param command the task to execute
     * @throws RejectedExecutionException at discretion of
     *         {@code RejectedExecutionHandler}, if the task
     *         cannot be accepted for execution
     * @throws NullPointerException if {@code command} is null
     */
    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. * 若是當前的線程數小於核心線程池的大小,根據現有的線程做爲第一個Worker運行的線程, * 新建一個Worker,addWorker自動的檢查當前線程池的狀態和Worker的數量, * 防止線程池在不能添加線程的狀態下添加線程 * * 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. * 若是線程入隊成功,而後仍是要進行double-check的,由於線程池在入隊以後狀態是可能會發生變化的 * * 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. * * 若是task不能入隊(隊列滿了),這時候嘗試增長一個新線程,若是增長失敗那麼當前的線程池狀態變化了或者線程池已經滿了 * 而後拒絕task */ int c = ctl.get(); //當前的Worker的數量小於核心線程池大小時,新建一個Worker。 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))//recheck防止線程池狀態的突變,若是突變,那麼將reject線程,防止workQueue中增長新線程 reject(command); else if (workerCountOf(recheck) == 0)//上下兩個操做都有addWorker的操做,可是若是在workQueue.offer的時候Worker變爲0, //那麼將沒有Worker執行新的task,因此增長一個Worker. addWorker(null, false); } //若是workQueue滿了,那麼這時候可能還沒到線程池的maxnum,因此嘗試增長一個Worker else if (!addWorker(command, false)) reject(command);//若是Worker數量到達上限,那麼就拒絕此線程 } 複製代碼

這裏須要明確幾個概念ide

  • Worker和Task的區別,Worker是當前線程池中的線程,而task雖然是runnable,可是並無真正執行,只是被Worker調用了run方法,後面會看到這部分的實現。
  • maximumPoolSize和corePoolSize的區別:這個概念很重要,maximumPoolSize爲線程池最大容量,也就是說線程池最多能起多少Worker。corePoolSize是核心線程池的大小,當corePoolSize滿了時,同時workQueue full(ArrayBolckQueue是可能滿的) 那麼此時容許新建Worker去處理workQueue中的Task,可是不能超過maximumPoolSize。超過corePoolSize以外的線程會在空閒超時後終止。

核心方法:addWorker Worker的增長和Task的獲取以及終止都是在此方法中實現的,也就是這一個方法裏面包含了不少東西。在addWorker方法中提到了Status的概念,Status是線程池的核心概念,這裏咱們先看一段關於status的註釋:oop

/**
     * 首先ctl是一個原子量,同時它裏面包含了兩個field,一個是workerCount,另外一個是runState
     * workerCount表示當前有效的線程數,也就是Worker的數量
     * runState表示當前線程池的狀態
     * The main pool control state, ctl, is an atomic integer packing
     * two conceptual fields
     *   workerCount, indicating the effective number of threads
     *   runState,    indicating whether running, shutting down etc
     * 
     * 二者是怎麼結合的呢?首先workerCount是佔據着一個atomic integer的後29位的,而狀態佔據了前3位
     * 因此,workerCount上限是(2^29)-1。
     * In order to pack them into one int, we limit workerCount to
     * (2^29)-1 (about 500 million) threads rather than (2^31)-1 (2
     * billion) otherwise representable. If this is ever an issue in
     * the future, the variable can be changed to be an AtomicLong,
     * and the shift/mask constants below adjusted. But until the need
     * arises, this code is a bit faster and simpler using an int.
     *
     * The workerCount is the number of workers that have been
     * permitted to start and not permitted to stop.  The value may be
     * transiently different from the actual number of live threads,
     * for example when a ThreadFactory fails to create a thread when
     * asked, and when exiting threads are still performing
     * bookkeeping before terminating. The user-visible pool size is
     * reported as the current size of the workers set.
     *
     * runState是整個線程池的運行生命週期,有以下取值:
     *  1. RUNNING:能夠新加線程,同時能夠處理queue中的線程。
     *  2. SHUTDOWN:不增長新線程,可是處理queue中的線程。
     *  3.STOP 不增長新線程,同時不處理queue中的線程。
     *  4.TIDYING 全部的線程都終止了(queue中),同時workerCount爲0,那麼此時進入TIDYING
     *  5.terminated()方法結束,變爲TERMINATED
     * The runState provides the main lifecyle control, taking on values:
     *
     *   RUNNING:  Accept new tasks and process queued tasks
     *   SHUTDOWN: Don't accept new tasks, but process queued tasks * STOP: Don't accept new tasks, don't process queued tasks, * and interrupt in-progress tasks * TIDYING: All tasks have terminated, workerCount is zero, * the thread transitioning to state TIDYING * will run the terminated() hook method * TERMINATED: terminated() has completed * * The numerical order among these values matters, to allow * ordered comparisons. The runState monotonically increases over * time, but need not hit each state. The transitions are: * 狀態的轉化主要是: * RUNNING -> SHUTDOWN(調用shutdown()) * On invocation of shutdown(), perhaps implicitly in finalize() * (RUNNING or SHUTDOWN) -> STOP(調用shutdownNow()) * On invocation of shutdownNow() * SHUTDOWN -> TIDYING(queue和pool均empty) * When both queue and pool are empty * STOP -> TIDYING(pool empty,此時queue已經爲empty) * When pool is empty * TIDYING -> TERMINATED(調用terminated()) * When the terminated() hook method has completed * * Threads waiting in awaitTermination() will return when the * state reaches TERMINATED. * * Detecting the transition from SHUTDOWN to TIDYING is less * straightforward than you'd like because the queue may become
     * empty after non-empty and vice versa during SHUTDOWN state, but
     * we can only terminate if, after seeing that it is empty, we see
     * that workerCount is 0 (which sometimes entails a recheck -- see
     * below).
     */
複製代碼

下面是狀態的代碼:學習

//利用ctl來保證當前線程池的狀態和當前的線程的數量。ps:低29位爲線程池容量,高3位爲線程狀態。
    private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
    //設定偏移量
    private static final int COUNT_BITS = Integer.SIZE - 3;
    //肯定最大的容量2^29-1
    private static final int CAPACITY   = (1 << COUNT_BITS) - 1;
    //幾個狀態,用Integer的高三位表示
    // runState is stored in the high-order bits
    //111
    private static final int RUNNING    = -1 << COUNT_BITS;
    //000
    private static final int SHUTDOWN   =  0 << COUNT_BITS;
    //001
    private static final int STOP       =  1 << COUNT_BITS;
    //010
    private static final int TIDYING    =  2 << COUNT_BITS;
    //011
    private static final int TERMINATED =  3 << COUNT_BITS;
    //獲取線程池狀態,取前三位
    // Packing and unpacking ctl
    private static int runStateOf(int c)     { return c & ~CAPACITY; }
    //獲取當前正在工做的worker,主要是取後面29位
    private static int workerCountOf(int c)  { return c & CAPACITY; }
    //獲取ctl
    private static int ctlOf(int rs, int wc) { return rs | wc; }
複製代碼

接下來貼上addWorker方法看看:

/**
     * Checks if a new worker can be added with respect to current
     * pool state and the given bound (either core or maximum). If so,
     * the worker count is adjusted accordingly, and, if possible, a
     * new worker is created and started running firstTask as its
     * first task. This method returns false if the pool is stopped or
     * eligible to shut down. It also returns false if the thread
     * factory fails to create a thread when asked, which requires a
     * backout of workerCount, and a recheck for termination, in case
     * the existence of this worker was holding up termination.
     *
     * @param firstTask the task the new thread should run first (or
     * null if none). Workers are created with an initial first task
     * (in method execute()) to bypass queuing when there are fewer
     * than corePoolSize threads (in which case we always start one),
     * or when the queue is full (in which case we must bypass queue).
     * Initially idle threads are usually created via
     * prestartCoreThread or to replace other dying workers.
     *
     * @param core if true use corePoolSize as bound, else
     * maximumPoolSize. (A boolean indicator is used here rather than a
     * value to ensure reads of fresh values after checking other pool
     * state).
     * @return true if successful
     */
    private boolean addWorker(Runnable firstTask, boolean core) {
        retry:
        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);
            // Check if queue empty only if necessary.
            /**
             * rs!=Shutdown || fistTask!=null || workCount.isEmpty
             * 若是當前的線程池的狀態>SHUTDOWN 那麼拒絕Worker的add 若是=SHUTDOWN
             * 那麼此時不能新加入不爲null的Task,若是在WorkCount爲empty的時候不能加入任何類型的Worker,
             * 若是不爲empty能夠加入task爲null的Worker,增長消費的Worker
             */
            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;
                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
            }
        }

        Worker w = new Worker(firstTask);
        Thread t = w.thread;

        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 c = ctl.get();
            int rs = runStateOf(c);
            /**
             * rs!=SHUTDOWN ||firstTask!=null
             * 
             * 一樣檢測當rs>SHUTDOWN時直接拒絕減少Wc,同時Terminate,若是爲SHUTDOWN同時firstTask不爲null的時候也要Terminate
             */
            if (t == null ||
                (rs >= SHUTDOWN &&
                 ! (rs == SHUTDOWN &&
                    firstTask == null))) {
                decrementWorkerCount();
                tryTerminate();
                return false;
            }

            workers.add(w);

            int s = workers.size();
            if (s > largestPoolSize)
                largestPoolSize = s;
        } finally {
            mainLock.unlock();
        }

        t.start();
        // It is possible (but unlikely) for a thread to have been
        // added to workers, but not yet started, during transition to
        // STOP, which could result in a rare missed interrupt,
        // because Thread.interrupt is not guaranteed to have any effect
        // on a non-yet-started Thread (see Thread#interrupt).
        //Stop或線程Interrupt的時候要停止全部的運行的Worker
        if (runStateOf(ctl.get()) == STOP && ! t.isInterrupted())
            t.interrupt();
        return true;
    }
複製代碼

addWorker中首先進行了一次線程池狀態的檢測:

int c = ctl.get();
            int rs = runStateOf(c);

            // Check if queue empty only if necessary.
            //判斷當前線程池的狀態是否是已經shutdown,若是shutdown了拒絕線程加入
            //(rs!=SHUTDOWN || first!=null || workQueue.isEmpty())
            //若是rs不爲SHUTDOWN,此時狀態是STOP、TIDYING或TERMINATED,因此此時要拒絕請求
            //若是此時狀態爲SHUTDOWN,而傳入一個不爲null的線程,那麼須要拒絕
            //若是狀態爲SHUTDOWN,同時隊列中已經沒任務了,那麼拒絕掉
            if (rs >= SHUTDOWN &&
                ! (rs == SHUTDOWN &&
                   firstTask == null &&
                   ! workQueue.isEmpty()))
                return false;
複製代碼

實際上是比較難懂的,主要在線程池狀態判斷條件這裏:

  1. 若是是runing,那麼跳過if。
  2. 若是rs>=SHUTDOWN,同時不等於SHUTDOWN,即爲SHUTDOWN以上的狀態,那麼不接受新線程。
  3. 若是rs>=SHUTDOWN,同時等於SHUTDOWN,同時first!=null,那麼拒絕新線程,若是first==null,那麼多是新增長線程消耗Queue中的線程。可是同時還要檢測workQueue是否isEmpty(),若是爲Empty,那麼隊列已空,不須要增長消耗線程,若是隊列沒有空那麼運行增長first=null的Worker。

從這裏是能夠看出一些策略的

  • 首先,在rs>SHUTDOWN時,拒絕一切線程的增長,由於STOP是會終止全部的線程,同時移除Queue中全部的待執行的線程的,因此也不須要增長first=null的Worker了
  • 其次,在SHUTDOWN狀態時,是不能增長first!=null的Worker的,同時即便first=null,可是此時Queue爲Empty也是不容許增長Worker的,SHUTDOWN下增長的Worker主要用於消耗Queue中的任務

SHUTDOWN狀態時,是不容許向workQueue中增長線程的,isRunning(c) && workQueue.offer(command) 每次在offer以前都要作狀態檢測,也就是線程池狀態變爲>=SHUTDOWN時不容許新線程進入線程池了

for (;;) {
                int wc = workerCountOf(c);
                //若是當前的數量超過了CAPACITY,或者超過了corePoolSize和maximumPoolSize(試core而定)
                if (wc >= CAPACITY ||
                    wc >= (core ? corePoolSize : maximumPoolSize))
                    return false;
                //CAS嘗試增長線程數,若是失敗,證實有競爭,那麼從新到retry。
                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
            }
複製代碼

這段代碼作了一個兼容,主要是沒有到corePoolSize 或maximumPoolSize上限時,那麼容許添加線程,CAS增長Worker的數量後,跳出循環。 接下來實例化Worker,實例化Worker實際上是很關鍵的,後面會說。 由於workers是HashSet線程不安全的,那麼此時須要加鎖,因此mainLock.lock(); 以後從新檢查線程池的狀態,若是狀態不正確,那麼減少Worker的數量,爲何tryTerminate()目前不大清楚。若是狀態正常,那麼添加Worker到workers。最後:

if (runStateOf(ctl.get()) == STOP && ! t.isInterrupted())
            t.interrupt();
複製代碼

註釋說的很清楚,爲了能及時的中斷此Worker,由於線程存在未Start的狀況,此時是不能響應中斷的,若是此時status變爲STOP,則不能中斷線程。此處用做中斷線程之用。 接下來咱們看Worker的方法:

/**
         * Creates with given first task and thread from ThreadFactory.
         * @param firstTask the first task (null if none)
         */
        Worker(Runnable firstTask) {
            this.firstTask = firstTask;
            this.thread = getThreadFactory().newThread(this);
        }
複製代碼

這裏能夠看出Worker是對firstTask的包裝,而且Worker自己就是Runnable的,看上去真心很煩。 經過ThreadFactory爲Worker本身構建一個線程。 由於Worker是Runnable類型的,因此是有run方法的,上面也看到了會調用t.start() 其實就是執行了run方法:

/** Delegates main run loop to outer runWorker  */
        public void run() {
            runWorker(this);
        }
複製代碼

調用了runWorker:

/**
     * Main worker run loop.  Repeatedly gets tasks from queue and
     * executes them, while coping with a number of issues:
     * 1 Worker可能仍是執行一個初始化的task——firstTask。
     *    可是有時也不須要這個初始化的task(能夠爲null),只要pool在運行,就會
     *   經過getTask從隊列中獲取Task,若是返回null,那麼worker退出。
     *   另外一種就是external拋出異常致使worker退出。
     * 1. We may start out with an initial task, in which case we
     * don't need to get the first one. Otherwise, as long as pool is * running, we get tasks from getTask. If it returns null then the * worker exits due to changed pool state or configuration * parameters. Other exits result from exception throws in * external code, in which case completedAbruptly holds, which * usually leads processWorkerExit to replace this thread. * * * 2 在運行任何task以前,都須要對worker加鎖來防止other pool中斷worker。 * clearInterruptsForTaskRun保證除了線程池stop,那麼現場都沒有中斷標誌 * 2. Before running any task, the lock is acquired to prevent * other pool interrupts while the task is executing, and * clearInterruptsForTaskRun called to ensure that unless pool is * stopping, this thread does not have its interrupt set. * * 3. Each task run is preceded by a call to beforeExecute, which * might throw an exception, in which case we cause thread to die * (breaking loop with completedAbruptly true) without processing * the task. * * 4. Assuming beforeExecute completes normally, we run the task, * gathering any of its thrown exceptions to send to * afterExecute. We separately handle RuntimeException, Error * (both of which the specs guarantee that we trap) and arbitrary * Throwables. Because we cannot rethrow Throwables within * Runnable.run, we wrap them within Errors on the way out (to the * thread's UncaughtExceptionHandler).  Any thrown exception also
     * conservatively causes thread to die.
     *
     * 5. After task.run completes, we call afterExecute, which may
     * also throw an exception, which will also cause thread to
     * die. According to JLS Sec 14.20, this exception is the one that
     * will be in effect even if task.run throws.
     *
     * The net effect of the exception mechanics is that afterExecute
     * and the thread's UncaughtExceptionHandler have as accurate * information as we can provide about any problems encountered by * user code. * * @param w the worker */ final void runWorker(Worker w) { Runnable task = w.firstTask; w.firstTask = null; //標識線程是否是異常終止的 boolean completedAbruptly = true; try { //task不爲null狀況是初始化worker時,若是task爲null,則去隊列中取線程--->getTask() while (task != null || (task = getTask()) != null) { w.lock(); //獲取woker的鎖,防止線程被其餘線程中斷 clearInterruptsForTaskRun();//清楚全部中斷標記 try { beforeExecute(w.thread, task);//線程開始執行以前執行此方法,能夠實現Worker未執行退出,本類中未實現 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);//線程執行後執行,能夠實現標識Worker異常中斷的功能,本類中未實現 } } finally { task = null;//運行過的task標null w.completedTasks++; w.unlock(); } } completedAbruptly = false; } finally { //處理worker退出的邏輯 processWorkerExit(w, completedAbruptly); } } 複製代碼

從上面代碼能夠看出,execute的Task是被「包裝 」了一層,線程啓動時是內部調用了Task的run方法。 接下來全部的核心集中在getTask()方法上:

/**
     * Performs blocking or timed wait for a task, depending on
     * current configuration settings, or returns null if this worker
     * must exit because of any of:
     * 1. There are more than maximumPoolSize workers (due to
     *    a call to setMaximumPoolSize).
     * 2. The pool is stopped.
     * 3. The pool is shutdown and the queue is empty.
     * 4. This worker timed out waiting for a task, and timed-out
     *    workers are subject to termination (that is,
     *    {@code allowCoreThreadTimeOut || workerCount > corePoolSize})
     *    both before and after the timed wait.
     *
     * @return task, or null if the worker must exit, in which case
     *         workerCount is decremented
     *         
     *         
     *  隊列中獲取線程
     */
    private Runnable getTask() {
        boolean timedOut = false; // Did the last poll() time out?

        retry:
        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);

            // Check if queue empty only if necessary.
            //當前狀態爲>stop時,不處理workQueue中的任務,同時減少worker的數量因此返回null,若是爲shutdown 同時workQueue已經empty了,一樣減少worker數量並返回null
            if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
                decrementWorkerCount();
                return null;
            }

            boolean timed;      // Are workers subject to culling?

            for (;;) {
                int wc = workerCountOf(c);
                timed = allowCoreThreadTimeOut || wc > corePoolSize;

                if (wc <= maximumPoolSize && ! (timedOut && timed))
                    break;
                if (compareAndDecrementWorkerCount(c))
                    return null;
                c = ctl.get();  // Re-read ctl
                if (runStateOf(c) != rs)
                    continue retry;
                // else CAS failed due to workerCount change; retry inner loop
            }

            try {
                Runnable r = timed ?
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                    workQueue.take();
                if (r != null)
                    return r;
                timedOut = true;
            } catch (InterruptedException retry) {
                timedOut = false;
            }
        }
    }
複製代碼

這段代碼十分關鍵,首先看幾個局部變量: boolean timedOut = false; 主要是判斷後面的poll是否要超時 boolean timed; 主要是標識着當前Worker超時是否要退出。wc > corePoolSize時須要減少空閒的Worker數,那麼timed爲true,可是wc <= corePoolSize時,不能減少核心線程數timed爲false。 timedOut初始爲false,若是timed爲true那麼使用poll取線程。若是正常返回,那麼返回取到的task。若是超時,證實worker空閒,同時worker超過了corePoolSize,須要刪除。返回r=null。則 timedOut = true。此時循環到wc <= maximumPoolSize && ! (timedOut && timed)時,減少worker數,並返回null,致使worker退出。若是線程數<= corePoolSize,那麼此時調用 workQueue.take(),沒有線程獲取到時將一直阻塞,知道獲取到線程或者中斷,關於中斷後面Shutdown的時候會說。

至此線程執行過程就分析完了

關於終止線程池

我我的認爲,若是想了解明白線程池,那麼就必定要理解好各個狀態之間的轉換,想理解轉換,線程池的終止機制是很好的一個途徑。對於關閉線程池主要有兩個方法shutdown()和shutdownNow(): 首先從shutdown()方法開始:

/**
     * Initiates an orderly shutdown in which previously submitted
     * tasks are executed, but no new tasks will be accepted.
     * Invocation has no additional effect if already shut down.
     *
     * <p>This method does not wait for previously submitted tasks to
     * complete execution.  Use {@link #awaitTermination awaitTermination}
     * to do that.
     *
     * @throws SecurityException {@inheritDoc}
     */
    public void shutdown() {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            //判斷是否能夠操做目標線程
            checkShutdownAccess();
            //設置線程池狀態爲SHUTDOWN,此處以後,線程池中不會增長新Task
            advanceRunState(SHUTDOWN);
            //中斷全部的空閒線程
            interruptIdleWorkers();
            onShutdown(); // hook for ScheduledThreadPoolExecutor
        } finally {
            mainLock.unlock();
        }
        //轉到Terminate
        tryTerminate();
    }
複製代碼

shutdown作了幾件事: 1. 檢查是否能操做目標線程 2. 將線程池狀態轉爲SHUTDOWN 3. 中斷全部空閒線程

這裏就引起了一個問題,什麼是空閒線程? 這須要接着看看interruptIdleWorkers是怎麼回事。

private void interruptIdleWorkers(boolean onlyOne) {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        //這裏的意圖很簡單,遍歷workers 對全部worker作中斷處理。
        // w.tryLock()對Worker加鎖,這保證了正在運行執行Task的Worker不會被中斷,那麼能中斷哪些線程呢?
        try {
            for (Worker w : workers) {
                Thread t = w.thread;
                if (!t.isInterrupted() && w.tryLock()) {
                    try {
                        t.interrupt();
                    } catch (SecurityException ignore) {
                    } finally {
                        w.unlock();
                    }
                }
                if (onlyOne)
                    break;
            }
        } finally {
            mainLock.unlock();
        }
    }
複製代碼

這裏主要是爲了中斷worker,可是中斷以前須要先獲取鎖,這就意味着正在運行的Worker不能中斷。可是上面的代碼有w.tryLock(),那麼獲取不到鎖就不會中斷,shutdown的Interrupt只是對全部的空閒Worker(正在從workQueue中取Task,此時Worker沒有加鎖)發送中斷信號。

while (task != null || (task = getTask()) != null) {
                w.lock();
                //獲取woker的鎖,防止線程被其餘線程中斷
                clearInterruptsForTaskRun();//清楚全部中斷標記
                try {
                    beforeExecute(w.thread, task);//線程開始執行以前執行此方法,能夠實現Worker未執行退出,本類中未實現
                    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);//線程執行後執行,能夠實現標識Worker異常中斷的功能,本類中未實現
                    }
                } finally {
                    task = null;//運行過的task標null
                    w.completedTasks++;
                    w.unlock();
                }
            }
複製代碼

在runWorker中,每個Worker getTask成功以後都要獲取Worker的鎖以後運行,也就是說運行中的Worker不會中斷。由於核心線程通常在空閒的時候會一直阻塞在獲取Task上,也只有中斷纔可能致使其退出。這些阻塞着的Worker就是空閒的線程(固然,非核心線程,而且阻塞的也是空閒線程)。在getTask方法中:

private Runnable getTask() {
        boolean timedOut = false; // Did the last poll() time out?

        retry:
        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);

            // Check if queue empty only if necessary.
            //當前狀態爲>stop時,不處理workQueue中的任務,同時減少worker的數量因此返回null,若是爲shutdown 同時workQueue已經empty了,一樣減少worker數量並返回null
            if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
                decrementWorkerCount();
                return null;
            }

            boolean timed;      // Are workers subject to culling?

            for (;;) {
                //allowCoreThreadTimeOu是判斷CoreThread是否會超時的,true爲會超時,false不會超時。默認爲false
                int wc = workerCountOf(c);
                timed = allowCoreThreadTimeOut || wc > corePoolSize;

                if (wc <= maximumPoolSize && ! (timedOut && timed))
                    break;
                if (compareAndDecrementWorkerCount(c))
                    return null;
                c = ctl.get();  // Re-read ctl
                if (runStateOf(c) != rs)
                    continue retry;
                // else CAS failed due to workerCount change; retry inner loop
            }

            try {
                Runnable r = timed ?
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                    workQueue.take();
                if (r != null)
                    return r;
                timedOut = true;
            } catch (InterruptedException retry) {
                timedOut = false;
            }
        }
    }
複製代碼

會有兩階段的Worker:

  • 剛進入getTask(),還沒進行狀態判斷。
  • block在poll或者take上的Worker。

當調用ShutDown方法時,首先設置了線程池的狀態爲ShutDown,此時1階段的worker進入到狀態判斷時會返回null,此時Worker退出。 由於getTask的時候是不加鎖的,因此在shutdown時能夠調用worker.Interrupt.此時會中斷退出,Loop到狀態判斷時,同時workQueue爲empty。那麼拋出中斷異常,致使從新Loop,在檢測線程池狀態時,Worker退出。若是workQueue不爲null就不會退出,此處有些疑問,由於沒有看見中斷標誌位清除的邏輯,那麼這裏就會不停的循環直到workQueue爲Empty退出。 這裏也能看出來SHUTDOWN只是清除一些空閒Worker,而且拒絕新Task加入,對於workQueue中的線程仍是繼續處理的。 對於shutdown中獲取mainLock而addWorker中也作了mainLock的獲取,這麼作主要是由於Works是HashSet類型的,是線程不安全的,咱們也看到在addWorker後面也是對線程池狀態作了判斷,將Worker添加和中斷邏輯分離開。 接下來作了tryTerminate()操做,這操做是進行了後面狀態的轉換,在shutdownNow後面說。 接下來看看shutdownNow:

/**
     * Attempts to stop all actively executing tasks, halts the
     * processing of waiting tasks, and returns a list of the tasks
     * that were awaiting execution. These tasks are drained (removed)
     * from the task queue upon return from this method.
     *
     * <p>This method does not wait for actively executing tasks to
     * terminate.  Use {@link #awaitTermination awaitTermination} to
     * do that.
     *
     * <p>There are no guarantees beyond best-effort attempts to stop
     * processing actively executing tasks.  This implementation
     * cancels tasks via {@link Thread#interrupt}, so any task that
     * fails to respond to interrupts may never terminate.
     *
     * @throws SecurityException {@inheritDoc}
     */
    public List<Runnable> shutdownNow() {
        List<Runnable> tasks;
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            checkShutdownAccess();
            advanceRunState(STOP);
            interruptWorkers();
            tasks = drainQueue();
        } finally {
            mainLock.unlock();
        }
        tryTerminate();
        return tasks;
    }
複製代碼

shutdownNow和shutdown代碼相似,可是實現卻很不相同。首先是設置線程池狀態爲STOP,前面的代碼咱們能夠看到,是對SHUTDOWN有一些額外的判斷邏輯,可是對於>=STOP,基本都是reject,STOP也是比SHUTDOWN更加嚴格的一種狀態。此時不會有新Worker加入,全部剛執行完一個線程後去GetTask的Worker都會退出。 以後調用interruptWorkers:

/**
     * Interrupts all threads, even if active. Ignores SecurityExceptions
     * (in which case some threads may remain uninterrupted).
     */
    private void interruptWorkers() {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            for (Worker w : workers) {
                try {
                    w.thread.interrupt();
                } catch (SecurityException ignore) {
                }
            }
        } finally {
            mainLock.unlock();
        }
    }
複製代碼

這裏能夠看出來,此方法目的是中斷全部的Worker,而不是像shutdown中那樣只中斷空閒線程。這樣體現了STOP的特色,中斷全部線程,同時workQueue中的Task也不會執行了。因此接下來drainQueue:

/**
     * Drains the task queue into a new list, normally using
     * drainTo. But if the queue is a DelayQueue or any other kind of
     * queue for which poll or drainTo may fail to remove some
     * elements, it deletes them one by one.
     */
    private List<Runnable> drainQueue() {
        BlockingQueue<Runnable> q = workQueue;
        List<Runnable> taskList = new ArrayList<Runnable>();
        q.drainTo(taskList);
        if (!q.isEmpty()) {
            for (Runnable r : q.toArray(new Runnable[0])) {
                if (q.remove(r))
                    taskList.add(r);
            }
        }
        return taskList;
    }
複製代碼

獲取全部沒有執行的Task,而且返回。 這也體現了STOP的特色: 拒絕全部新Task的加入,同時中斷全部線程,WorkerQueue中沒有執行的線程所有拋棄。因此此時Pool是空的,WorkerQueue也是空的。 這以後就是進行到TIDYING和TERMINATED的轉化了:

/**
     * Transitions to TERMINATED state if either (SHUTDOWN and pool
     * and queue empty) or (STOP and pool empty).  If otherwise
     * eligible to terminate but workerCount is nonzero, interrupts an
     * idle worker to ensure that shutdown signals propagate. This
     * method must be called following any action that might make
     * termination possible -- reducing worker count or removing tasks
     * from the queue during shutdown. The method is non-private to
     * allow access from ScheduledThreadPoolExecutor.
     */
    final void tryTerminate() {
        for (;;) {
            int c = ctl.get();
            if (isRunning(c) ||
                runStateAtLeast(c, TIDYING) ||
                (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
                return;
            if (workerCountOf(c) != 0) { // Eligible to terminate
                interruptIdleWorkers(ONLY_ONE);
                return;
            }

            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
                if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
                    try {
                        terminated();
                    } finally {
                        ctl.set(ctlOf(TERMINATED, 0));
                        termination.signalAll();
                    }
                    return;
                }
            } finally {
                mainLock.unlock();
            }
            // else retry on failed CAS
        }
    }
複製代碼

上面的代碼其實頗有意思有幾種狀態是不能轉化到TIDYING的:

  1. RUNNING狀態
  2. TIDYING或TERMINATED
  3. SHUTDOWN狀態,可是workQueue不爲空

也說明了兩點: 1. SHUTDOWN想轉化爲TIDYING,須要workQueue爲空,同時workerCount爲02. STOP轉化爲TIDYING,須要workerCount爲0

若是知足上面的條件(通常必定時間後都會知足的),那麼CAS成TIDYING,TIDYING也只是個過分狀態,最終會轉化爲TERMINATED。

至此,ThreadPoolExecutor一些核心思想就介紹完了,想分析清楚實在是不容易,對於ThreadPoolExecutor我仍是有些不懂地方,以上只是我對源碼的片面的看法,若是有不正確之處,但願各位大佬指出

共同進步,學習分享

歡迎你們關注個人公衆號【風平浪靜如碼】,海量Java相關文章,學習資料都會在裏面更新,整理的資料也會放在裏面。

以爲寫的還不錯的就點個贊,加個關注唄!點關注,不迷路,持續更新!!!

相關文章
相關標籤/搜索