Java線程池源碼及原理

時間 2019-12-08

標籤 java 線程源碼原理欄目 Java 简体版

原文原文鏈接

目錄java

1 說明

下面若是有貼出源碼，對應的源碼是JDK8
主要的源碼類
java.util.concurrent.ThreadPoolExecutor、
java.util.concurrent.ThreadPoolExecutor.Worker
java.util.concurrent.AbstractExecutorService源碼分析

1.1類繼承圖

2 線程池的狀態

3 源碼分析

3.1完整的線程池構造方法

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

3.2 ctl

內部有重要的成員變量ctl，類型是AtomicInteger，低29位表示線程池中線程數，經過高3位表示線程池的運行狀態
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()方法已經執行完成this

3.3 任務的執行

execute --> addWorker --> Thread.start --> (Thread.run) --> runTask --> getTask.net

3.3.1 execute(Runnable command)

大體分三個步驟
一、當前運行的線程數量是否小於corePoolSize，直接嘗試addWorker()
二、往阻塞隊列裏面放入Runnable任務
三、若是隊列已經滿了，直接嘗試addWorker()線程

3.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方法會被執行code

3.3.3 runWorker(Worker w)

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

3.3.4 getTask()

直接貼源碼了blog

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

代碼比較簡單，把任務封裝成一個既實現Runnable, 也實現Future 的接口，這個時候就能夠調用execute()進行實現了接口