Java中 ThreadPoolExecutor 類學習筆記

 

Java中 ThreadPoolExecutor 類學習筆記

一、定義

Java中的線程池。ThreadPoolExecutor類是接口Executor的實現類。以下圖所示： 

二、做用

線程池解決兩個不一樣的問題：因爲每一個任務的調用開銷減小，它們一般在執行大量異步任務時提供改進的性能，而且它們提供了一種限制和管理資源（包括執行一個任務。 每一個ThreadPoolExecutor還維護一些基本統計信息，例如已完成任務的數量。

三、建立類解釋

ThreadPoolExecutor extends AbstractExecutorService。

構造方法：

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 ： 線程池的核心線程數，即便空閒時仍保留在池中的線程數，除非設置 allowCoreThreadTimeOut.
2. 　　maximumPoolSize : 池中容許的最大線程數
3.        keepAliveTime :  線程的存活時間。當線程池裏的線程數大於corePoolSize時，若是等了keepAliveTime時長尚未任務可執行，則線程退出。
4.        unit  : 指定keepAliveTime的單位
5.        workQueue :  一個阻塞隊列，提交的任務將會被放到這個隊列裏
6.    　threadFactory :  執行程序建立新線程時使用的工廠
7.        handler : 執行被阻止時使用的處理程序，由於達到線程限制和隊列容量 

4.線程池執行流程

任務被提交到線程池，會先判斷當前線程數量是否小於corePoolSize，若是小於則建立線程來執行提交的任務，不然將任務放入workQueue隊列，若是workQueue滿了，則判斷當前線程數量是否小於maximumPoolSize,若是小於則建立線程執行任務，不然就會調用handler，以表示線程池拒絕接收任務。

 5.線程池的幾個主要方法分析

5.1 主方法：ThreadPoolExector的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.
         *
         * 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.
         *
         * 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.
         */
　　　　　　
　　　　　//1 當前運行的線程數量小於核心線程數量，直接將任務加入worker啓動運行。
        int c = ctl.get();
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
　

　　/**
　　2 運行線程數量大於核心線程數量時，上面的if分支針對大於corePoolSize，而且緩存隊列加入任務操做成功的狀況。
　　運行中而且將任務加入緩衝隊列成功，正常來講這樣已經完成了處理邏輯。
　　可是爲了保險起見，增長了狀態出現異常的確認判斷，若是狀態出現異常會繼續remove操做，若是執行true，則按照拒絕處理策略駁回任務；
　　*/

        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); }

　　/**
　　3 這裏針對運行線程數量超過了corePoolSize，而且緩存隊列也已經放滿的狀況。
　　注意第二個參數是false，能夠在下面addWorker方法看到，就是針對線程池最大線程數量maximumPoolSize的判斷。
　　*/

　　else if (!addWorker(command, false)) reject(command); 　　}
}

 5.2 關鍵方法：ThreadPoolExector的addWorker（增長工做線程）

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)) //建立非核心線程時，即core等於false。判斷當前線程數是否大於等於maximumPoolSize，若是大於等於則返回false 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
            }
        }

        boolean workerStarted = false;
        boolean workerAdded = false;
        Worker w = null;
        try {
            w = new Worker(firstTask); final Thread t = w.thread;//建立Worker對象，同時也會實例化一個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();
                        if (s > largestPoolSize)
                            largestPoolSize = s;
                        workerAdded = true;
                    }
                } finally {
                    mainLock.unlock();
                }
                if (workerAdded) {
                    t.start();
                    workerStarted = true;
                }
            }
        } finally {
            if (! workerStarted)
                addWorkerFailed(w);
        }
        return workerStarted;
    }

5.3 Worker中的runWorker方法，也是worker中的run方法主體。

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

/**
     * Main worker run loop.  Repeatedly gets tasks from queue and
     * executes them, while coping with a number of issues:
     *
     * 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. Before running any task, the lock is acquired to prevent
     * other pool interrupts while the task is executing, and then we
     * 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) {
        Thread wt = Thread.currentThread();
        Runnable task = w.firstTask;
        w.firstTask = null;
        w.unlock(); // allow interrupts
        boolean completedAbruptly = true;
        try {
            while (task != null || (task = getTask()) != null) {//線程調用runWoker，會while循環調用getTask方法從workerQueue裏讀取任務，而後執行任務
                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
                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();//只要getTask方法不返回null,此線程就不會退出。
                    } 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);
        }
 }

5.4 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, and if the queue is * non-empty, this worker is not the last thread in the pool. * * @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?

        for (;;) { int c = ctl.get(); int rs = runStateOf(c); // Check if queue empty only if necessary.
            if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) { decrementWorkerCount(); return null; } int wc = workerCountOf(c); // Are workers subject to culling?
            boolean timed = allowCoreThreadTimeOut || wc > corePoolSize; 　　　　　　　
　　　　　　　//若是運行線程數超過了最大線程數，可是緩存隊列已經空了，這時遞減worker數量。
　　　　　　　//若是有設置容許線程超時或者線程數量超過了核心線程數量，而且線程在規定時間內均未poll到任務且隊列爲空則遞減worker數量
 if ((wc > maximumPoolSize || (timed && timedOut)) && (wc > 1 || workQueue.isEmpty())) { if (compareAndDecrementWorkerCount(c)) return null; continue; } try { Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take(); if (r != null) return r; timedOut = true; } catch (InterruptedException retry) { timedOut = false; } } }