java線程池原理解析

五一假期大雄看了一本《java併發編程藝術》，瞭解了線程池的基本工做流程，居然發現線程池工做原理和互聯網公司運做模式十分類似。

線程池處理流程

原理解析

互聯網公司與線程池的關係

這裏用一個比喻來描述一下線程池，中間有一些名詞你可能不是太清楚，後邊源碼解析的部分會講到。

你能夠把線程池看做是一個研發部門，研發部門有不少程序員（Worker), 他們在一個大辦公室裏（HashSet workers)。程序員幹不完的需求（Runnable/Callable)放在需求池（workQueue）裏排隊。每一個研發部都配置有骨幹程序員數量（corePoolSize）和最大能容納的程序員數量（maximumPoolSize）。具體要作的任務就是產品的需求。

new 一個線程池至關於建立了一個研發部，建立研發部時須要指定骨幹程序員數量，最大能容納的程序員數量，需求池用哪一種（BlockingQueue）,若是忙不過來的需求怎麼給產品回覆（拒絕策略）等等內容。剛開始這個研發部一個程序員也沒有。

當產品給這個研發部提一個需求時（固然確定不會只提一個，他們會不斷的提需求。這裏以提一個需求爲例）

首先會看骨幹程序員招聘滿了沒。

若是沒滿，會招聘一個骨幹程序員，招聘進來就讓他不停的工做（很殘酷啊），幹完剛派過來的任務他會主動在需求池找下一個需求來作（好員工），若是需求池沒有需求了，他就中止工做了，而後研發部會把他裁掉，若是裁掉後發現骨幹程序員數量不夠了，就會再招聘一個程序員。裁掉後，要是骨幹程序員數量還夠就不招聘了。

若是骨幹程序員數量滿了，就看需求池滿沒滿，若是需求池沒滿，就把需求扔進需求池裏；若是需求池滿了，就看程序員數量有沒有達到上限，若是達到了，就對產品說，這個需求咱們作不了，沒資源；若是沒達到，就招聘一個程序員，招聘進來就讓他不停的工做，幹完剛派過來的需求他會主動到需求池找下一個任務來作，若是需求池沒有任務了，他就中止工做了，而後研發部會把他裁掉，若是裁掉後發現骨幹程序員數量不夠了，就會再招聘一個程序員。裁掉後，要是骨幹程序員數量還夠就不招聘了。

源碼解析

首先是worker（程序員）

Worker被裝在一個HashSet（workers）裏邊， 他是用來執行任務的，他們的職責就是不斷的從workQueue裏邊取任務，而後執行。當workQueue（需求池）裏邊拿不到任務，或者線程池達到特定狀態，worker就會從workers裏邊移走（被裁）。

下邊是Worker源碼，移除了非關鍵的東西

private final class Worker
    extends AbstractQueuedSynchronizer
    implements Runnable
{

    // 標識這個任務是在哪一個線程運行
    final Thread thread;
    Runnable firstTask;
    // 完成了幾個任務
    volatile long completedTasks;

    Worker(Runnable firstTask) {
        // 阻止中斷，知道runWorker執行
        setState(-1); // inhibit interrupts until runWorker
        this.firstTask = firstTask;
        // 直接用你提供的線程工廠搞個線程出來
        this.thread = getThreadFactory().newThread(this);
    }

    // 調用ThreadPoolExecutor裏邊的runWorker方法
    public void run() {
        runWorker(this);
    }

    // 如下這些是AQS相關的東西

    // 0表明沒有加鎖
    // 1表明加鎖了
    protected boolean isHeldExclusively() {
        return getState() != 0;
    }

    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); }
    public boolean isLocked() { return isHeldExclusively(); }

    void interruptIfStarted() {
        Thread t;
        if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
            try {
                t.interrupt();
            } catch (SecurityException ignore) {
            }
        }
    }
}

Worker實現了Runnable接口，因此他是個任務，有run方法；同時有繼承了AQS，因此他也是一把鎖。

下邊是提交任務的過程

提交任務有submit和execute, submit就是首先將Callable或者Runnable包裝成FutureTask，而後調用execute, 因此核心是分析execute

public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();
    // 這個c裏邊有兩個信息，一個是如今有多少worker, 另外一個是如今線程池的狀態是啥
    // workerCountOf方法就是從裏邊提取 worker的數量的
    int c = ctl.get();
    if (workerCountOf(c) < corePoolSize) { // 當前worker的數量比須要的核心線程數少
        // 加worker去執行,加成功就完事了，也就是說只要worker比核心線程數少，就會建立worker
        // 無論如今核心線程是否在工做，也無論workQueue是否是滿的
        // addWorker的第二個參數表示是否是要加核心線程（或者叫核心worker)
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    // 當前worker達到或超過了核心線程數，或者加worker失敗了，纔會走下邊的流程
    // worker已經比核心線程數多了

    // 若是 線程池沒有shutdown的話 
    // 就嘗試將任務加到workQueue裏邊，工做隊列入隊成功的話再往裏邊走
    if (isRunning(c) && workQueue.offer(command)) {
        int recheck = ctl.get();
        if (! isRunning(recheck) && remove(command))
            // 再次檢查狀態若是線程池要停了，那麼就拒絕任務，而且把worker從工做隊列扔掉
            reject(command);
        else if (workerCountOf(recheck) == 0)
            // 若是沒有worker的話(說明沒加進去，這種場景我沒想到是什麼狀況)，加一個worker
            addWorker(null, false);
        // 其餘狀況，丟到工做隊列就不用管了，等着worker去處理
    }
    // 若是隊列滿了加失敗了，或者線程池狀態不知足了，就嘗試加普通worker(非核心線程)
    else if (!addWorker(command, false))
        // 加失敗了就拒絕任務
        // 失敗一方面多是worker數量已經達到你的給的maximumPoolSize
        // 另外一方面，多是檢查到線程池的狀態不對了
        reject(command);
}

能夠發現execute方法就是完成了上邊說的「線程池處理流程」這個圖裏描述的過程。 大雄看到這裏還有幾個疑問，一個是Woker是如何建立並加入workers的，一個是worker是如何啓動的，再就是worker是如何運行的

生活還要繼續

private boolean addWorker(Runnable firstTask, boolean core) {
    retry:
    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);

        // 作一些校驗，線程池的狀態要知足必定條件
        // 並且得提交任務過來，再就是workQueue不能是空的
        if (rs >= SHUTDOWN &&
            ! (rs == SHUTDOWN &&
                firstTask == null &&
                ! workQueue.isEmpty()))
            return false;

        for (;;) {
            int wc = workerCountOf(c);
            // 看你是要建立核心worker仍是普通worker
            // 核心看超沒超過corePoolSize, 普通看超沒超過maximumPoolSize
            if (wc >= CAPACITY ||
                wc >= (core ? corePoolSize : maximumPoolSize))
                return false;
            if (compareAndIncrementWorkerCount(c))
            // 增長worker數量失敗就在來
                break retry;
            c = ctl.get();  // Re-read ctl
            if (runStateOf(c) != rs)
                // 中途線程池狀態發生變化了
                continue retry;
        }
    }

    boolean workerStarted = false;
    boolean workerAdded = false;
    Worker w = null;
    try {
        // worker就是這麼建立的
        w = new Worker(firstTask);
        final Thread t = w.thread;
        if (t != null) {
            final ReentrantLock mainLock = this.mainLock;
            // 加worker是要加全局鎖的
            mainLock.lock();
            try {
                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) {
                // worker是在這裏啓動的
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        if (! workerStarted)
            addWorkerFailed(w);
    }
    return workerStarted;
}

這段代碼解決了 Woker是如何建立並加入workers的以及worker是如何啓動的的問題。

addWorker作的核心工做就是，建立worker, 啓動worker, 在建立以前還會作一些校驗。調用了worker裏邊線程的start後就要等待cpu調度執行worker的run方法了。

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

final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    w.unlock(); // allow interrupts
    boolean completedAbruptly = true;
    try {
        // task是建立worker帶進去的任務，會先執行他，而後從workQueue裏邊取
        // 若是沒有的話跳出去
        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， 默認是空的
                beforeExecute(wt, task);
                Throwable thrown = null;
                try {
                    // 這個跟咱們平時理解的Runnable還不同，能夠體會下，他這個run就是一個普通的方法
                    // 他直接調run是要執行任務，線程的start只是把worker裏邊的那個run跑起來了
                    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 {
        // 從while跳出來代表沒有任務能夠執行了
        processWorkerExit(w, completedAbruptly);
    }
}

這個也比較容易，就是不斷的從workQueue取任務，執行，直到沒任務了跳出來。接下來就是worker如何被銷燬的問題了

private void processWorkerExit(Worker w, boolean completedAbruptly) {
    if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
        decrementWorkerCount();

    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        completedTaskCount += w.completedTasks;
        // 移除掉worker（裁人）
        workers.remove(w);
    } finally {
        mainLock.unlock();
    }

    tryTerminate();

    int c = ctl.get();
    if (runStateLessThan(c, STOP)) {
        if (!completedAbruptly) {
            int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
            if (min == 0 && ! workQueue.isEmpty())
                min = 1;
            // 比核心線程數多的話，執行完的Worker直接移除就好
            if (workerCountOf(c) >= min)
                return; // replacement not needed
        }
        // 小於核心線程數就會再加個Worker, 讓他繼續等待接收任務（招人）
        addWorker(null, false);
    }
}

直接從workers裏邊移除worker, 移除後若是worker數量比核心線程數還少，就再加個worker, 不然不加。

一些體會

看源碼必定不要過度糾結細節，就像這個線程池，我看網上不少文章去算那幾個位運算的十進制數，感受是在浪費時間，沒有抓住重點。

固然這也不是絕對的（彷佛說的矛盾了），一些細節的設計仍是很是精妙值得學習的。仍是這個位運算，爲何只用一個int表示線程池狀態和worker的數量呢。

要多多聯想，仍是這個位運算，他是否是和讀寫鎖用一個int既表示寫狀態又表示讀狀態十分類似。Worker繼承AQS，是否能讓你想起AQS的種種。

總之，我的以爲第一遍看是必定不能沉溺於細節的，他會讓你迷惘和喪失信心；第二遍、第三遍能夠關注一下細節，感覺大師級的設計的美妙之處。固然筆者僅僅粗略看了一遍（逃~）

最後

大雄五一假期閱讀了《java併發編程藝術》這本書，整理了一本gitbook筆記（還沒寫完），須要的同窗能夠掃描文末二維碼關注「大雄和你一塊兒學編程」公衆號，後臺回覆我愛java領取。這本gitbook還沒完全完成，因此可能還有些小錯誤。將來會大約每兩天推送其中的一篇文章。

以下是這本gitbook的目錄截圖