1. 程式人生 > >Java執行緒池原始碼及原理

Java執行緒池原始碼及原理

目錄

  • 1 說明
    • 1.1類繼承圖
  • 2 執行緒池的狀態
  • 3 原始碼分析
    • 3.1完整的執行緒池構造方法
    • 3.2 ctl
    • 3.3 任務的執行
      • 3.3.1 execute(Runnable command)
      • 3.3.2 addWorker(Runnable firstTask, boolean core)
      • 3.3.3 runWorker(Worker w)
      • 3.3.4 getTask()
    • 4 任務執行,帶返回值的
  • 5 參考資料

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
1、RUNNING:-1 << COUNT_BITS,即高3位為111,該狀態的執行緒池會接收新任務;
2、SHUTDOWN: 0 << COUNT_BITS,即高3位為000,該狀態的執行緒池不會接收新任務;
3、STOP : 1 << COUNT_BITS,即高3位為001;
4、TIDYING : 2 << COUNT_BITS,即高3位為010, 所有的任務都已經終止;
5、TERMINATED: 3 << COUNT_BITS,即高3位為011, terminated()方法已經執行完成

3.3 任務的執行

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

3.3.1 execute(Runnable command)

大致分三個步驟
1、當前執行的執行緒數量是否小於corePoolSize,直接嘗試addWorker()
2、往阻塞佇列裡面放入Runnable任務
3、如果佇列已經滿了,直接嘗試addWorker()

3.3.2 addWorker(Runnable firstTask, boolean core)

1、前置判斷執行緒池的狀態
2、通過CAS操作讓ctl加1,表示執行執行緒數增加1個
3、構造一個Worker w,這裡要特別注意構造方法裡面的這行程式碼,this.thread = getThreadFactory().newThread(this),可以看到構造方法內,有一個Thread物件,其使用了ThreadFactory構造了一個新的執行緒,並且執行緒的runable是worker本身。
4、執行w.thread.start(),也就是說,當該執行緒被執行時,Worker中的run方法會被執行

3.3.3 runWorker(Worker w)

通過迴圈呼叫getTask()獲取要執行的任務task
beforeExecute
task.run()
afterExecute

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

程式碼比較簡單,把任務封裝成一個既實現Runnable, 也實現Future的介面,這個時候就可以呼叫execute()進行實現了

5 參考資料

https://blog.csdn.net/programmer_at/article/details/79799267
https://blog.csdn.net/liuzhixiong_521/article/details/8785