面試官：你分析過線程池源碼嗎？

線程池源碼也是面試經常被提問到的點，我會將全局源碼做一分析，然后告訴你面試考啥，怎么答。

為什么要用線程池?

簡潔的答兩點就行。

1. 降低系統資源消耗。

2. 提高線程可控性。

如何創建使用線程池?

JDK8提供了五種創建線程池的方法：

1.創建一個定長線程池，可控制線程最大並發數，超出的線程會在隊列中等待。

public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>());
}

 

2.(JDK8新增)會根據所需的並發數來動態創建和關閉線程。能夠合理的使用CPU進行對任務進行並發操作，所以適合使用在很耗時的任務。

注意返回的是ForkJoinPool對象。

 

public static ExecutorService newWorkStealingPool(int parallelism) {
    return new ForkJoinPool
        (parallelism,
         ForkJoinPool.defaultForkJoinWorkerThreadFactory,
         null, true);
}

 

什么是ForkJoinPool：

public ForkJoinPool(int parallelism,
                        ForkJoinWorkerThreadFactory factory,
                        UncaughtExceptionHandler handler,
                        boolean asyncMode) {
        this(checkParallelism(parallelism),
             checkFactory(factory),
             handler,
             asyncMode ? FIFO_QUEUE : LIFO_QUEUE,
             "ForkJoinPool-" + nextPoolId() + "-worker-");
        checkPermission();
    }

 

使用一個無限隊列來保存需要執行的任務，可以傳入線程的數量；不傳入，則默認使用當前計算機中可用的cpu數量；使用分治法來解決問題，使用fork()和join()來進行調用。

3.創建一個可緩存的線程池,可靈活回收空閑線程，若無可回收，則新建線程。

public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue<Runnable>());
}

 

4.創建一個單線程的線程池。

public static ExecutorService newSingleThreadExecutor() {
    return new FinalizableDelegatedExecutorService
        (new ThreadPoolExecutor(1, 1,
                                0L, TimeUnit.MILLISECONDS,
                                new LinkedBlockingQueue<Runnable>()));
}

 

5.創建一個定長線程池，支持定時及周期性任務執行。

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
    return new ScheduledThreadPoolExecutor(corePoolSize);
}

 

上層源碼結構分析

Executor結構:

Executor

一個運行新任務的簡單接口

 public interface Executor { 2 3 void execute(Runnable command); 4 5 } 

ExecutorService

擴展了Executor接口。添加了一些用來管理執行器生命周期和任務生命周期的方法

AbstractExecutorService

對ExecutorService接口的抽象類實現。不是我們分析的重點。

ThreadPoolExecutor

Java線程池的核心實現。

ThreadPoolExecutor源碼分析

屬性解釋

// AtomicInteger是原子類  ctlOf()返回值為RUNNING；
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
// 高3位表示線程狀態
private static final int COUNT_BITS = Integer.SIZE - 3;
// 低29位表示workerCount容量
private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

// runState is stored in the high-order bits
// 能接收任務且能處理阻塞隊列中的任務
private static final int RUNNING    = -1 << COUNT_BITS;
// 不能接收新任務，但可以處理隊列中的任務。
private static final int SHUTDOWN   =  0 << COUNT_BITS;
// 不接收新任務，不處理隊列任務。
private static final int STOP       =  1 << COUNT_BITS;
// 所有任務都終止
private static final int TIDYING    =  2 << COUNT_BITS;
// 什么都不做
private static final int TERMINATED =  3 << COUNT_BITS;

// 存放任務的阻塞隊列
private final BlockingQueue<Runnable> workQueue;

 

值的注意的是狀態值越大線程越不活躍。

線程池狀態的轉換模型：

構造器

public ThreadPoolExecutor(int corePoolSize,//線程池初始啟動時線程的數量
                          int maximumPoolSize,//最大線程數量
                          long keepAliveTime,//空閑線程多久關閉?
                          TimeUnit unit,// 計時單位
                          BlockingQueue<Runnable> workQueue,//放任務的阻塞隊列
                          ThreadFactory threadFactory,//線程工廠
                          RejectedExecutionHandler handler// 拒絕策略) {
    if (corePoolSize < 0 ||
        maximumPoolSize <= 0 ||
        maximumPoolSize < corePoolSize ||
        keepAliveTime < 0)
        throw new IllegalArgumentException();
    if (workQueue == null || threadFactory == null || handler == null)
        throw new NullPointerException();
    this.acc = System.getSecurityManager() == null ?
            null :
            AccessController.getContext();
    this.corePoolSize = corePoolSize;
    this.maximumPoolSize = maximumPoolSize;
    this.workQueue = workQueue;
    this.keepAliveTime = unit.toNanos(keepAliveTime);
    this.threadFactory = threadFactory;
    this.handler = handler;
}

 

在向線程池提交任務時，會通過兩個方法：execute和submit。

本文着重講解execute方法。submit方法放在下次和Future、Callable一起分析。

execute方法:

public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();
    // clt記錄着runState和workerCount
    int c = ctl.get();
    //workerCountOf方法取出低29位的值，表示當前活動的線程數
    //然后拿線程數和 核心線程數做比較
    if (workerCountOf(c) < corePoolSize) {
        // 如果活動線程數<核心線程數
        // 添加到
        //addWorker中的第二個參數表示限制添加線程的數量是根據corePoolSize來判斷還是maximumPoolSize來判斷
        if (addWorker(command, true))
            // 如果成功則返回
            return;
        // 如果失敗則重新獲取 runState和 workerCount
        c = ctl.get();
    }
    // 如果當前線程池是運行狀態並且任務添加到隊列成功
    if (isRunning(c) && workQueue.offer(command)) {
        // 重新獲取 runState和 workerCount
        int recheck = ctl.get();
        // 如果不是運行狀態並且 
        if (! isRunning(recheck) && remove(command))
            reject(command);
        else if (workerCountOf(recheck) == 0)
            //第一個參數為null，表示在線程池中創建一個線程，但不去啟動
            // 第二個參數為false，將線程池的有限線程數量的上限設置為maximumPoolSize
            addWorker(null, false);
    }
    //再次調用addWorker方法，但第二個參數傳入為false，將線程池的有限線程數量的上限設置為maximumPoolSize
    else if (!addWorker(command, false))
        //如果失敗則拒絕該任務
        reject(command);
}

 

總結一下它的工作流程：

1. 當workerCount < corePoolSize，創建線程執行任務。

2. 當workerCount >= corePoolSize&&阻塞隊列workQueue未滿，把新的任務放入阻塞隊列。

3. 當workQueue已滿，並且workerCount >= corePoolSize，並且workerCount < maximumPoolSize，創建線程執行任務。

4. 當workQueue已滿，workerCount >= maximumPoolSize，采取拒絕策略,默認拒絕策略是直接拋異常。

 

通過上面的execute方法可以看到，最主要的邏輯還是在addWorker方法中實現的，那我們就看下這個方法：

addWorker方法

主要工作是在線程池中創建一個新的線程並執行

參數定義：

• firstTask the task the new thread should run first (or null if none). (指定新增線程執行的第一個任務或者不執行任務)

• core if true use corePoolSize as bound, else maximumPoolSize.(core如果為true則使用corePoolSize綁定，否則為maximumPoolSize。 （此處使用布爾指示符而不是值，以確保在檢查其他狀態后讀取新值）。)

 

private boolean addWorker(Runnable firstTask, boolean core) {
    retry:
    for (;;) {

        int c = ctl.get();
        //  獲取運行狀態
        int rs = runStateOf(c);

        // Check if queue empty only if necessary.
        // 如果狀態值 >= SHUTDOWN (不接新任務&不處理隊列任務)
        // 並且 如果 ！(rs為SHUTDOWN 且 firsTask為空 且 阻塞隊列不為空)
        if (rs >= SHUTDOWN &&
            ! (rs == SHUTDOWN &&
               firstTask == null &&
               ! workQueue.isEmpty()))
            // 返回false
            return false;

        for (;;) {
            //獲取線程數wc
            int wc = workerCountOf(c);
            // 如果wc大與容量 || core如果為true表示根據corePoolSize來比較,否則為maximumPoolSize
            if (wc >= CAPACITY ||
                wc >= (core ? corePoolSize : maximumPoolSize))
                return false;
            // 增加workerCount（原子操作）
            if (compareAndIncrementWorkerCount(c))
                // 如果增加成功，則跳出
                break retry;
            // wc增加失敗，則再次獲取runState
            c = ctl.get();  // Re-read ctl
            // 如果當前的運行狀態不等於rs，說明狀態已被改變，返回重新執行
            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 {
        // 根據firstTask來創建Worker對象
        w = new Worker(firstTask);
        // 根據worker創建一個線程
        final Thread t = w.thread;
        if (t != null) {
            // new一個鎖
            final ReentrantLock mainLock = this.mainLock;
            // 加鎖
            mainLock.lock();
            try {
                // Recheck while holding lock.
                // Back out on ThreadFactory failure or if
                // shut down before lock acquired.
                // 獲取runState
                int rs = runStateOf(ctl.get());
                // 如果rs小於SHUTDOWN(處於運行)或者(rs=SHUTDOWN && firstTask == null)
                // firstTask == null證明只新建線程而不執行任務
                if (rs < SHUTDOWN ||
                    (rs == SHUTDOWN && firstTask == null)) {
                    // 如果t活着就拋異常
                    if (t.isAlive()) // precheck that t is startable
                        throw new IllegalThreadStateException();
                    // 否則加入worker(HashSet)
                    //workers包含池中的所有工作線程。僅在持有mainLock時訪問。
                    workers.add(w);
                    // 獲取工作線程數量
                    int s = workers.size();
                    //largestPoolSize記錄着線程池中出現過的最大線程數量
                    if (s > largestPoolSize)
                        // 如果 s比它還要大，則將s賦值給它
                        largestPoolSize = s;
                    // worker的添加工作狀態改為true    
                    workerAdded = true;
                }
            } finally {
                mainLock.unlock();
            }
            // 如果worker的添加工作完成
            if (workerAdded) {
                // 啟動線程
                t.start();
                // 修改線程啟動狀態
                workerStarted = true;
            }
        }
    } finally {
        if (! workerStarted)
            addWorkerFailed(w);
    }
    // 返回線啟動狀態
    return workerStarted;

 

為什么需要持有mainLock？

因為workers是HashSet類型的，不能保證線程安全。

那w = new Worker(firstTask);如何理解呢

Worker.java

private final class Worker
    extends AbstractQueuedSynchronizer
    implements Runnable

 

可以看到它繼承了AQS並發框架還實現了Runnable。證明它還是一個線程任務類。那我們調用t.start()事實上就是調用了該類重寫的run方法。

Worker為什么不使用ReentrantLock來實現呢？

tryAcquire方法它是不允許重入的，而ReentrantLock是允許重入的。對於線程來說，如果線程正在執行是不允許其它鎖重入進來的。

線程只需要兩個狀態，一個是獨占鎖，表明正在執行任務；一個是不加鎖，表明是空閑狀態。

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

run方法又調用了runWorker方法：

final void runWorker(Worker w) {
    // 拿到當前線程
    Thread wt = Thread.currentThread();
    // 拿到當前任務
    Runnable task = w.firstTask;
    // 將Worker.firstTask置空 並且釋放鎖
    w.firstTask = null;
    w.unlock(); // allow interrupts
    boolean completedAbruptly = true;
    try {
        // 如果task或者getTask不為空，則一直循環
        while (task != null || (task = getTask()) != null) {
            // 加鎖
            w.lock();
            // If pool is stopping, ensure thread is interrupted;
            // if not, ensure thread is not interrupted.  This
            // requires a recheck in second case to deal with
            // shutdownNow race while clearing interrupt
            //  return ctl.get() >= stop 
            // 如果線程池狀態>=STOP 或者 (線程中斷且線程池狀態>=STOP)且當前線程沒有中斷
            // 其實就是保證兩點：
            // 1. 線程池沒有停止
            // 2. 保證線程沒有中斷
            if ((runStateAtLeast(ctl.get(), STOP) ||
                 (Thread.interrupted() &&
                  runStateAtLeast(ctl.get(), STOP))) &&
                !wt.isInterrupted())
                // 中斷當前線程
                wt.interrupt();
            try {
                // 空方法
                beforeExecute(wt, task);
                Throwable thrown = null;
                try {
                    // 執行run方法(Runable對象)
                    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置空， 完成任務++， 釋放鎖
                task = null;
                w.completedTasks++;
                w.unlock();
            }
        }
        completedAbruptly = false;
    } finally {
        // 退出工作
        processWorkerExit(w, completedAbruptly);
    }

 

總結一下runWorker方法的執行過程：

1. while循環中，不斷地通過getTask()方法從workerQueue中獲取任務

2. 如果線程池正在停止，則中斷線程。否則調用3.

3. 調用task.run()執行任務；

4. 如果task為null則跳出循環，執行processWorkerExit()方法，銷毀線程workers.remove(w);

這個流程圖非常經典：

除此之外，ThreadPoolExector還提供了tryAcquire、tryRelease、shutdown、shutdownNow、tryTerminate、等涉及的一系列線程狀態更改的方法有興趣可以自己研究。大體思路是一樣的，這里不做介紹。

Worker為什么不使用ReentrantLock來實現呢？

tryAcquire方法它是不允許重入的，而ReentrantLock是允許重入的。對於線程來說，如果線程正在執行是不允許其它鎖重入進來的。

線程只需要兩個狀態，一個是獨占鎖，表明正在執行任務；一個是不加鎖，表明是空閑狀態。

在runWorker方法中，為什么要在執行任務的時候對每個工作線程都加鎖呢？

shutdown方法與getTask方法存在競態條件.(這里不做深入，建議自己深入研究，對它比較熟悉的面試官一般會問)

高頻考點

1. 創建線程池的五個方法。

2. 線程池的五個狀態

3. execute執行過程。

4. runWorker執行過程。(把兩個流程圖記下，理解后說個大該就行。)

5. 比較深入的問題就是我在文中插入的問題。

6. …期望大家能在評論區補充。

聲明：圖片來源於網絡，侵刪。