[JUC源碼剖析]__ThreadPoolExecutor類

ThreadPoolExecutor構造方法

工做中不少時候都會用到線程池，可是線程池內部是怎麼實現的呢

先看一下ThreadPoolExecutor類的構造方法

public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler);

corePoolSize: 核心線程池大小，當線程池中的線程數小於corePoolSize時，每提交一個任務，都會新起一個線程來處理任務。線程會不斷的從workQueue中取出任務執行。線程通常狀況下即便空閒，也不會回收，除非設置了allowCoreThreadTimeOut參數

workQueue：工做隊列，當線程數達到了corePoolSize後，後續提交的任務就會插入到workQueue中

maximumPoolSize：線程池最大線程數， 當workQueue滿了以後，線程池就會啓動新的線程來處理任務，可是整個線程池的線程數最大不會超過maximumPoolSize

keepAliveTime和unit：非core線程的最大空閒時間和時間單位

threadFactory： 線程工廠，線程池會使用線程工廠來建立線程

handler：飽和策略，當線程池的線程達到maximumPoolSize且workQueue滿了後，會使用handler處理新提交的任務

注意：不少人會搞錯corePoolSize，maximumPoolSize，workQueue之間的關係，認爲是core線程滿了以後，會直接建立新的線程處理任務而不用插入到workQueue中。其實是workQueue滿了以後纔會建立新的線程，總的線程數量不超過maximumPoolSize

這裏是一個線程池使用demo

static void demo()throws Exception{
        ExecutorService executorService =  new ThreadPoolExecutor(1, 1,
                0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>());
        Future<String> future = executorService.submit(() -> {
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "hello world";
        });

        System.out.println(future.get());;
    }

線程池是如何建立線程的？

AbstractExecutorService類

public <T> Future<T> submit(Callable<T> task) {
    if (task == null) throw new NullPointerException();
    RunnableFuture<T> ftask = newTaskFor(task);
    execute(ftask); //執行任務
    return ftask;
}
protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
    return new FutureTask<T>(runnable, value);
}

因爲submit方法返回的是提供Future，因此提交任務的時候實際上提交的是一個RunnableFuture接口的實現類FutureTask。而execure(ftask)則是任務執行的核心

public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();
    int c = ctl.get();

    // 當worker數小於corePoolSize時則建立worker
    if (workerCountOf(c) < corePoolSize) {
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    
    // 當worker大於等於corePoolSize且線程池是運行中時，則嘗試插入任務到workerQueue中
    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);
    }
    // 當線程數大於等於coreSize且workerQueue滿了時，則再次嘗試增長worker
    else if (!addWorker(command, false))
        reject(command);
}

代碼中的worker能夠理解爲線程池中執行任務的線程，能夠看到corePoolSize，workQueue間的關係是：

1. 當worker數小於corePoolSize時則建立worker
2. 當worker大於等於corePoolSize且線程池是運行中時，則嘗試插入任務到workerQueue中
3. 當線程數大於等於coreSize且workerQueue滿了時，則再次嘗試增長worker

這裏有個比較有意思的設計就是 private final AtomicInteger ctl;這個變量。它是一個32位的整數類型,高3位表明了線程池的狀態，低29位表明線程池中活躍的線程數。

爲何要把兩個變量合併到一個變量中呢？個人理解就是這樣設計就能夠在同一個cas操做中保證在設置數量的時候，狀態是不變的。若是分開成兩個變量，除非加更重的鎖，不然在增長數量的過程當中，狀態是有可能改變的。

那麼問題來了：maximumPoolSize的做用是怎麼體現的呢？
先看看private boolean 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.
        if (rs >= SHUTDOWN &&
            ! (rs == SHUTDOWN &&
                firstTask == null &&
                ! workQueue.isEmpty()))
            return false;

        for (;;) {
            int wc = workerCountOf(c);
            //核心worker大於corePoolSize，非核心線程大於maximumPoolSize則增長失敗
            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
        }
    }

    boolean workerStarted = false;
    boolean workerAdded = false;
    Worker w = null;
    try {
        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();
                    if (s > largestPoolSize)
                        largestPoolSize = s;
                    workerAdded = true;
                }
            } finally {
                mainLock.unlock();
            }
            if (workerAdded) {
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        if (! workerStarted)
            addWorkerFailed(w);
    }
    return workerStarted;
}

第17行能夠看到在增長worker時，是會校驗當前的worker數量的
在方法的第一個嵌套自旋中能夠看到，裏面有不少的狀態判斷和worker數量判斷，當全部判斷成功時會經過compareAndIncrementWorkerCount方法去修改ctl變量的worker數量

在JUC包中，做者大量的使用了自旋和CAS操做來代替鎖操做，這種操做屬於樂觀鎖

上面提到了線程池狀態，而線程池存在五個狀態，且各個狀態間可以轉化

五個狀態：
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

狀態間的轉化
RUNNING -> SHUTDOWN
        On invocation of shutdown(), perhaps implicitly in finalize()
(RUNNING or SHUTDOWN) -> STOP
        On invocation of shutdownNow()
SHUTDOWN -> TIDYING
        When both queue and pool are empty
STOP -> TIDYING
        When pool is empty
TIDYING -> TERMINATED
        When the terminated() hook method has completed

Worker是怎麼從Queue中消費任務的？

先看看Worker類的

private final class Worker
        extends AbstractQueuedSynchronizer
        implements Runnable{

        private static final long serialVersionUID = 6138294804551838833L;


        final Thread thread;
        Runnable firstTask;
        volatile long completedTasks;

        Worker(Runnable firstTask) {
            setState(-1); // inhibit interrupts until runWorker
            this.firstTask = firstTask;
            this.thread = getThreadFactory().newThread(this);
        }

        public void run() {
            runWorker(this);
        }
    //
}

Worker自己就是一個Runnable，它包含了一個Thread字段用於執行認爲。線程池中線程的數量其實就是Worker的數量。而Worker中的線程最終執行的就是裏面的runWorker方法

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) {
                w.lock();
                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);
        }
    }

能夠看到有一個while循環會不斷的獲取任務執行，當獲取到task後，接下來就會執行task.run方法。
那麼假如隊列爲空時，core線程不是會繼續保存在線程池中，非core線程會等待一段時間後再銷燬嗎？這個邏輯是怎麼實現的？答案就在getTask()方法中

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;

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

能夠看到getTask方法會根據線程數是否大於corePoolSize來或者allowCoreThreadTimeOut是否爲true來決定從workQueue中獲取任務時可否超時返回。
當容許超時返回，則超時後getTask會返回null，且在runWorker中當getTask返回null時則會調用processWorkerExit方法終止當前worker的線程。

當不容許超時返回時，則會一直阻塞在workQueue.take()中

到這裏爲止就搞懂這3個問題了

1. 線程池的構造參數是如何起做用的？
2. 線程池是如何建立線程的？
3. Worker是怎麼從Queue中消費任務的？