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的目錄截圖

筆記目錄截圖

歡迎關注「大雄和你一起學編程」公眾號