Java多線程系列--「JUC線程池」03之線程池原理(二)

時間 2019-12-09

標籤 java 多線程系列 juc 線程原理欄目 Java 简体版

原文原文鏈接

概要

在前面一章"Java多線程系列--「JUC線程池」02之線程池原理(一)"中介紹了線程池的數據結構，本章會經過分析線程池的源碼，對線程池進行說明。內容包括：
線程池示例
 參考代碼(基於JDK1.7.0_40)
線程池源碼分析
 (一) 建立「線程池」
(二) 添加任務到「線程池」
(三) 關閉「線程池」html

轉載請註明出處：http://www.cnblogs.com/skywang12345/p/3509954.htmljava

線程池示例

在分析線程池以前，先看一個簡單的線程池示例。數據結構

 1 import java.util.concurrent.Executors;
 2 import java.util.concurrent.ExecutorService;
 3 
 4 public class ThreadPoolDemo1 {
 5 
 6     public static void main(String[] args) {
 7         // 建立一個可重用固定線程數的線程池
 8         ExecutorService pool = Executors.newFixedThreadPool(2);
 9         // 建立實現了Runnable接口對象，Thread對象固然也實現了Runnable接口
10         Thread ta = new MyThread();
11         Thread tb = new MyThread();
12         Thread tc = new MyThread();
13         Thread td = new MyThread();
14         Thread te = new MyThread();
15         // 將線程放入池中進行執行
16         pool.execute(ta);
17         pool.execute(tb);
18         pool.execute(tc);
19         pool.execute(td);
20         pool.execute(te);
21         // 關閉線程池
22         pool.shutdown();
23     }
24 }
25 
26 class MyThread extends Thread {
27 
28     @Override
29     public void run() {
30         System.out.println(Thread.currentThread().getName()+ " is running.");
31     }
32 }

運行結果：多線程

pool-1-thread-1 is running.
pool-1-thread-2 is running.
pool-1-thread-1 is running.
pool-1-thread-2 is running.
pool-1-thread-1 is running.

示例中，包括了線程池的建立，將任務添加到線程池中，關閉線程池這3個主要的步驟。稍後，咱們會從這3個方面來分析ThreadPoolExecutor。ide

參考代碼(基於JDK1.7.0_40)

Executors完整源碼函數

View Code

ThreadPoolExecutor完整源碼oop

View Code

線程池源碼分析

(一) 建立「線程池」源碼分析

下面以newFixedThreadPool()介紹線程池的建立過程。this

1. newFixedThreadPool()spa

newFixedThreadPool()在Executors.java中定義，源碼以下：

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

說明：newFixedThreadPool(int nThreads)的做用是建立一個線程池，線程池的容量是nThreads。
newFixedThreadPool()在調用ThreadPoolExecutor()時，會傳遞一個LinkedBlockingQueue()對象，而LinkedBlockingQueue是單向鏈表實現的阻塞隊列。在線程池中，就是經過該阻塞隊列來實現"當線程池中任務數量超過容許的任務數量時，部分任務會阻塞等待"。
關於LinkedBlockingQueue的實現細節，讀者能夠參考"Java多線程系列--「JUC集合」08之 LinkedBlockingQueue"。

2. ThreadPoolExecutor()

ThreadPoolExecutor()在ThreadPoolExecutor.java中定義，源碼以下：

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue) {
    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
         Executors.defaultThreadFactory(), defaultHandler);
}

說明：該函數其實是調用ThreadPoolExecutor的另一個構造函數。該函數的源碼以下：

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.corePoolSize = corePoolSize;
    // 最大池大小
    this.maximumPoolSize = maximumPoolSize;
    // 線程池的等待隊列
    this.workQueue = workQueue;
    this.keepAliveTime = unit.toNanos(keepAliveTime);
    // 線程工廠對象
    this.threadFactory = threadFactory;
    // 拒絕策略的句柄
    this.handler = handler;
}

說明：在ThreadPoolExecutor()的構造函數中，進行的是初始化工做。
corePoolSize, maximumPoolSize, unit, keepAliveTime和workQueue這些變量的值是已知的，它們都是經過newFixedThreadPool()傳遞而來。下面看看threadFactory和handler對象。

2.1 ThreadFactory

線程池中的ThreadFactory是一個線程工廠，線程池建立線程都是經過線程工廠對象(threadFactory)來完成的。
上面所說的threadFactory對象，是經過 Executors.defaultThreadFactory()返回的。Executors.java中的defaultThreadFactory()源碼以下：

public static ThreadFactory defaultThreadFactory() {
    return new DefaultThreadFactory();
}

defaultThreadFactory()返回DefaultThreadFactory對象。Executors.java中的DefaultThreadFactory()源碼以下：

static class DefaultThreadFactory implements ThreadFactory {
    private static final AtomicInteger poolNumber = new AtomicInteger(1);
    private final ThreadGroup group;
    private final AtomicInteger threadNumber = new AtomicInteger(1);
    private final String namePrefix;

    DefaultThreadFactory() {
        SecurityManager s = System.getSecurityManager();
        group = (s != null) ? s.getThreadGroup() :
                              Thread.currentThread().getThreadGroup();
        namePrefix = "pool-" +
                      poolNumber.getAndIncrement() +
                     "-thread-";
    }

    // 提供建立線程的API。
    public Thread newThread(Runnable r) {
        // 線程對應的任務是Runnable對象r
        Thread t = new Thread(group, r,
                              namePrefix + threadNumber.getAndIncrement(),
                              0);
        // 設爲「非守護線程」
        if (t.isDaemon())
            t.setDaemon(false);
        // 將優先級設爲「Thread.NORM_PRIORITY」
        if (t.getPriority() != Thread.NORM_PRIORITY)
            t.setPriority(Thread.NORM_PRIORITY);
        return t;
    }
}

說明：ThreadFactory的做用就是提供建立線程的功能的線程工廠。
它是經過newThread()提供建立線程功能的，下面簡單說說newThread()。newThread()建立的線程對應的任務是Runnable對象，它建立的線程都是「非守護線程」並且「線程優先級都是Thread.NORM_PRIORITY」。

2.2 RejectedExecutionHandler

handler是ThreadPoolExecutor中拒絕策略的處理句柄。所謂拒絕策略，是指將任務添加到線程池中時，線程池拒絕該任務所採起的相應策略。
線程池默認會採用的是defaultHandler策略，即AbortPolicy策略。在AbortPolicy策略中，線程池拒絕任務時會拋出異常！
defaultHandler的定義以下：

private static final RejectedExecutionHandler defaultHandler = new AbortPolicy();

AbortPolicy的源碼以下：

public static class AbortPolicy implements RejectedExecutionHandler {
    public AbortPolicy() { }

    // 拋出異常
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        throw new RejectedExecutionException("Task " + r.toString() +
                                             " rejected from " +
                                             e.toString());
    }
}

(二) 添加任務到「線程池」

1. execute()

execute()定義在ThreadPoolExecutor.java中，源碼以下：

public void execute(Runnable command) {
    // 若是任務爲null，則拋出異常。
    if (command == null)
        throw new NullPointerException();
    // 獲取ctl對應的int值。該int值保存了"線程池中任務的數量"和"線程池狀態"信息
    int c = ctl.get();
    // 當線程池中的任務數量 < "核心池大小"時，即線程池中少於corePoolSize個任務。
    // 則經過addWorker(command, true)新建一個線程，並將任務(command)添加到該線程中；而後，啓動該線程從而執行任務。
    if (workerCountOf(c) < corePoolSize) {
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    // 當線程池中的任務數量 >= "核心池大小"時，
    // 並且，"線程池處於容許狀態"時，則嘗試將任務添加到阻塞隊列中。
    if (isRunning(c) && workQueue.offer(command)) {
        // 再次確認「線程池狀態」，若線程池異常終止了，則刪除任務；而後經過reject()執行相應的拒絕策略的內容。
        int recheck = ctl.get();
        if (! isRunning(recheck) && remove(command))
            reject(command);
        // 不然，若是"線程池中任務數量"爲0，則經過addWorker(null, false)嘗試新建一個線程，新建線程對應的任務爲null。
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    // 經過addWorker(command, false)新建一個線程，並將任務(command)添加到該線程中；而後，啓動該線程從而執行任務。
    // 若是addWorker(command, false)執行失敗，則經過reject()執行相應的拒絕策略的內容。
    else if (!addWorker(command, false))
        reject(command);
}

說明：execute()的做用是將任務添加到線程池中執行。它會分爲3種狀況進行處理：
  狀況1 -- 若是"線程池中任務數量" < "核心池大小"時，即線程池中少於corePoolSize個任務；此時就新建一個線程，並將該任務添加到線程中進行執行。
  狀況2 -- 若是"線程池中任務數量" >= "核心池大小"，而且"線程池是容許狀態"；此時，則將任務添加到阻塞隊列中阻塞等待。在該狀況下，會再次確認"線程池的狀態"，若是"第2次讀到的線程池狀態"和"第1次讀到的線程池狀態"不一樣，則從阻塞隊列中刪除該任務。
  狀況3 -- 非以上兩種狀況。在這種狀況下，嘗試新建一個線程，並將該任務添加到線程中進行執行。若是執行失敗，則經過reject()拒絕該任務。

2. addWorker()

addWorker()的源碼以下：

private boolean addWorker(Runnable firstTask, boolean core) {
    retry:
    // 更新"線程池狀態和計數"標記，即更新ctl。
    for (;;) {
        // 獲取ctl對應的int值。該int值保存了"線程池中任務的數量"和"線程池狀態"信息
        int c = ctl.get();
        // 獲取線程池狀態。
        int rs = runStateOf(c);

        // 有效性檢查
        if (rs >= SHUTDOWN &&
            ! (rs == SHUTDOWN &&
               firstTask == null &&
               ! workQueue.isEmpty()))
            return false;

        for (;;) {
            // 獲取線程池中任務的數量。
            int wc = workerCountOf(c);
            // 若是"線程池中任務的數量"超過限制，則返回false。
            if (wc >= CAPACITY ||
                wc >= (core ? corePoolSize : maximumPoolSize))
                return false;
            // 經過CAS函數將c的值+1。操做失敗的話，則退出循環。
            if (compareAndIncrementWorkerCount(c))
                break retry;
            c = ctl.get();  // Re-read ctl
            // 檢查"線程池狀態"，若是與以前的狀態不一樣，則從retry從新開始。
            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 {
        final ReentrantLock mainLock = this.mainLock;
        // 新建Worker，而且指定firstTask爲Worker的第一個任務。
        w = new Worker(firstTask);
        // 獲取Worker對應的線程。
        final Thread t = w.thread;
        if (t != null) {
            // 獲取鎖
            mainLock.lock();
            try {
                int c = ctl.get();
                int rs = runStateOf(c);

                // 再次確認"線程池狀態"
                if (rs < SHUTDOWN ||
                    (rs == SHUTDOWN && firstTask == null)) {
                    if (t.isAlive()) // precheck that t is startable
                        throw new IllegalThreadStateException();
                    // 將Worker對象(w)添加到"線程池的Worker集合(workers)"中
                    workers.add(w);
                    // 更新largestPoolSize
                    int s = workers.size();
                    if (s > largestPoolSize)
                        largestPoolSize = s;
                    workerAdded = true;
                }
            } finally {
                // 釋放鎖
                mainLock.unlock();
            }
            // 若是"成功將任務添加到線程池"中，則啓動任務所在的線程。 
            if (workerAdded) {
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        if (! workerStarted)
            addWorkerFailed(w);
    }
    // 返回任務是否啓動。
    return workerStarted;
}

說明：
addWorker(Runnable firstTask, boolean core) 的做用是將任務(firstTask)添加到線程池中，並啓動該任務。
core爲true的話，則以corePoolSize爲界限，若"線程池中已有任務數量>=corePoolSize"，則返回false；core爲false的話，則以maximumPoolSize爲界限，若"線程池中已有任務數量>=maximumPoolSize"，則返回false。
addWorker()會先經過for循環不斷嘗試更新ctl狀態，ctl記錄了"線程池中任務數量和線程池狀態"。
更新成功以後，再經過try模塊來將任務添加到線程池中，並啓動任務所在的線程。

從addWorker()中，咱們能清晰的發現：線程池在添加任務時，會建立任務對應的Worker對象；而一個Workder對象包含一個Thread對象。(01) 經過將Worker對象添加到"線程的workers集合"中，從而實現將任務添加到線程池中。 (02) 經過啓動Worker對應的Thread線程，則執行該任務。

3. submit()

補充說明一點，submit()實際上也是經過調用execute()實現的，源碼以下：

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

(三) 關閉「線程池」

shutdown()的源碼以下：

public void shutdown() {
    final ReentrantLock mainLock = this.mainLock;
    // 獲取鎖
    mainLock.lock();
    try {
        // 檢查終止線程池的「線程」是否有權限。
        checkShutdownAccess();
        // 設置線程池的狀態爲關閉狀態。
        advanceRunState(SHUTDOWN);
        // 中斷線程池中空閒的線程。
        interruptIdleWorkers();
        // 鉤子函數，在ThreadPoolExecutor中沒有任何動做。
        onShutdown(); // hook for ScheduledThreadPoolExecutor
    } finally {
        // 釋放鎖
        mainLock.unlock();
    }
    // 嘗試終止線程池
    tryTerminate();
}

說明：shutdown()的做用是關閉線程池。