十分鐘深刻理解Java線程池源碼及設計原理

時間 2019-11-07

標籤十分 10分深刻理解 java 線程源碼設計原理欄目 Java 简体版

原文原文鏈接

1、說明

下面若是有貼出源碼，對應的源碼是JDK8java

主要的源碼類源碼分析

java.util.concurrent.ThreadPoolExecutor、
java.util.concurrent.ThreadPoolExecutor.Worker
java.util.concurrent.AbstractExecutorServicethis

1. 類繼承圖

2、線程池的狀態

3、源碼分析

1. 完整的線程池構造方法

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

2. ctl

內部有重要的成員變量ctl，類型是AtomicInteger，低29位表示線程池中線程數，經過高3位表示線程池的運行狀態spa

COUNT_BITS的值是29線程

RUNNING：-1 << COUNT_BITS，即高3位爲111，該狀態的線程池會接收新任務；
SHUTDOWN： 0 << COUNT_BITS，即高3位爲000，該狀態的線程池不會接收新任務；
STOP ： 1 << COUNT_BITS，即高3位爲001；
TIDYING ： 2 << COUNT_BITS，即高3位爲010, 全部的任務都已經終止；
TERMINATED： 3 << COUNT_BITS，即高3位爲011, terminated()方法已經執行完成

3. 任務的執行

execute --> addWorker --> Thread.start --> (Thread.run) --> runTask --> getTaskcode

3.1execute(Runnable command)orm

大體分三個步驟cdn

當前運行的線程數量是否小於corePoolSize，直接嘗試addWorker()
往阻塞隊列裏面放入Runnable任務
若是隊列已經滿了，直接嘗試addWorker()

3.2 addWorker(Runnable firstTask, boolean core)對象

前置判斷線程池的狀態
經過CAS操做讓ctl加1，表示運行線程數增長1個
構造一個Worker w，這裏要特別注意構造方法裏面的這行代碼，this.thread = getThreadFactory().newThread(this)，能夠看到構造方法內，有一個Thread對象，其使用了ThreadFactory構造了一個新的線程，而且線程的runable是worker自己。
執行w.thread.start()，也就是說，當該線程被運行時，Worker中的run方法會被執行

3.3 runWorker(Worker w)blog

經過循環調用getTask()獲取要執行的任務task
beforeExecute
task.run()
afterExecute

3.4 getTask()

源碼以下：

private Runnable getTask() {
    Boolean timedOut = false;
    // 是否最後的 poll() 超時了？
    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);
        Boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
        // worker是否須要被淘汰
        if ((wc > maximumPoolSize || (timed && timedOut))
                    && (wc > 1 || workQueue.isEmpty())) {
            // 這裏會讓線程的數量記錄減，後面的return null，會致使runWorker沒有獲取到數據而讓run()方法走到盡頭，最終當前線程結束
            if (compareAndDecrementWorkerCount(c))
                            return null;
            continue;
        }
        try {
            // 若是須要回收一部分線程，那麼超時時間keepAliveTime後拿不到就數據就繼續循環調用，就能夠在下一次循環的時候進行線程結束回收了；不然一直阻塞下去
            Runnable r = timed ?
                            workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                            workQueue.take();
            if (r != null)
                            return r;
            timedOut = true;
        }
        catch (InterruptedException retry) {
            timedOut = false;
        }
    }
}
複製代碼

4、任務執行，帶返回值的

源碼以下：

public <T> Future<T> submit(Callable<T> task) {
    if (task == null) throw new NullPointerException();
    RunnableFuture<T> ftask = newTaskFor(task);
    execute(ftask);
    return ftask;
}
public Future<?> submit(Runnable task) {
    if (task == null) throw new NullPointerException();
    RunnableFuture<Void> ftask = newTaskFor(task, null);
    execute(ftask);
    return ftask;
}
複製代碼