ThreadPoolExecutor原理

時間 2020-07-20

標籤 threadpoolexecutor 原理简体版

原文原文鏈接

關注能夠查看更多粉絲專享blog~java

線程池狀態及轉換條件圖

飽和策略（當隊列滿而且線程個數達到maximunPoolSize後採起的策略）

AbortPolicy：拋出異常。
CallerRunsPolicy：使用調用者所在的線程來運行任務。
DiscardOldestPolicy：調用poll丟棄一個任務，執行當前任務。
DiscardPolicy：默默丟棄，不拋出異常。

Executors線程池類型

keepAliveTime線程空閒keepAliveTime後則回收。ide

newFixedThreadPool

// 核心線程和最大線程數同樣，隊列長度爲Integer.MAX_VALUE，keepAliveTime=0
public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>());
}

// 使用自定義線程建立工廠
public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>(),
                                  threadFactory);
}
複製代碼

newSingleThreadExecutor

// 核心線程和最大線程數都爲1，隊列長度爲Integer.MAX_VALUE，keepAliveTime=0
public static ExecutorService newSingleThreadExecutor() {
    return new FinalizableDelegatedExecutorService
        (new ThreadPoolExecutor(1, 1,
                                0L, TimeUnit.MILLISECONDS,
                                new LinkedBlockingQueue<Runnable>()));
}

// 使用自定義線程建立工廠
public static ExecutorService newSingleThreadExecutor(ThreadFactory threadFactory) {
    return new FinalizableDelegatedExecutorService
        (new ThreadPoolExecutor(1, 1,
                                0L, TimeUnit.MILLISECONDS,
                                new LinkedBlockingQueue<Runnable>(),
                                threadFactory));
}
複製代碼

newCachedThreadPool

// 初始核心線程數爲0，最大線程數爲Integer.MAX_VALUE，而且爲同步阻塞隊列，keepAliveTime=60
// 特殊之處在於，加入同步隊列的任務會立刻執行，同步隊列裏面最多隻有一個任務
public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue<Runnable>());
}

// 使用自定義線程建立工廠
public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) {
  return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue<Runnable>(),
                                  threadFactory);
}
複製代碼

newScheduledThreadPool

核心方法1：execute(Runnable command)

public void execute(Runnable command) {
	// 參數校驗
    if (command == null)
        throw new NullPointerException();
    // 獲取當前線程池狀態和線程個數變量組合值（高3位爲狀態，低29位爲線程數量）
    int c = ctl.get();
    // 若是當前線程個數小於核心線程數則增長核心線程運行任務
    if (workerCountOf(c) < corePoolSize) {
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    // 走到這一步說明線程數大於核心線程數，若是線程池處於RUNNING狀態，則添加到阻塞隊列
    if (isRunning(c) && workQueue.offer(command)) {
        int recheck = ctl.get();
        // 再次判斷狀態，期間可能會有其餘線程執行shutdown等操做改變狀態，若是不是RUNNING則移除任務，並執行拒絕策略
        if (! isRunning(recheck) && remove(command))
            reject(command);
        // 不然若是當前線程爲空，則添加一個線程
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    // 若是隊列滿，則新增線程，新增失敗則執行拒絕策略
    else if (!addWorker(command, false))
        reject(command);
}
複製代碼

核心方法2：addWorker(Runnable firstTask, boolean core)

private boolean addWorker(Runnable firstTask, boolean core) {
    retry:
    // 1.增長線程個數
    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);

        // 只在必要時檢查隊列是否爲空
        if (rs >= SHUTDOWN &&
            ! (rs == SHUTDOWN &&
               firstTask == null &&
               ! workQueue.isEmpty()))
            return false;

		// 循環CAS增長線程個數
        for (;;) {
        	// 獲取當前工做線程個數
            int wc = workerCountOf(c);
            // 若是線程個數超了則return false
            if (wc >= CAPACITY ||
                wc >= (core ? corePoolSize : maximumPoolSize))
                return false;
            // 增長工做線程個數
            if (compareAndIncrementWorkerCount(c))
                break retry;
            // 增長失敗，則查看線程池狀態是否變化了，若是發生變化則跳到外層循環從新嘗試獲取線程池狀態，不然內層循環從新CAS增長線程個數
            c = ctl.get();
            if (runStateOf(c) != rs)
                continue retry;
        }
    }

	// 2.添加到工做隊列
    boolean workerStarted = false;
    boolean workerAdded = false;
    Worker w = null;
    try {
    	// 構造worker，狀態設置爲-1禁止中斷，直到runWorker時修改成可中斷，剛構造的worker中斷無心義
    	// Worker(Runnable firstTask) {
        // setState(-1);
        // this.firstTask = firstTask;
        // this.thread = getThreadFactory().newThread(this);
        // }
        w = new Worker(firstTask);
        final Thread t = w.thread;
        if (t != null) {
        	// 獲取獨佔鎖，爲了實現同步workers同步，存在多個線程調用了線程池的execute
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
            	// 從新獲取線程池狀態
                int rs = runStateOf(ctl.get());
                // 若是狀態爲可執行任務狀態，或者爲shutdown
                if (rs < SHUTDOWN ||
                    (rs == SHUTDOWN && firstTask == null)) {
                    // 檢查線程是否能夠啓動，啓動過則報錯
                    // 方法isAlive() 的功能是判斷當前的線程是否處於活動狀態；活動狀態就是線程已經啓動還沒有終止，那麼這時候線程就是存活的，則返回true，不然則返回false；
                    if (t.isAlive())
                        throw new IllegalThreadStateException();
                    workers.add(w);
                    int s = workers.size();
                    if (s > largestPoolSize)
                        largestPoolSize = s;
                    workerAdded = true;
                }
            } finally {
            	// 釋放鎖
                mainLock.unlock();
            }
            // 若是添加成功則啓動線程
            if (workerAdded) {
                t.start();
                workerStarted = true;
            }
        }
    } finally {
    	// 添加失敗則從新獲取鎖，而後從移除當前worker，CAS workerCount - 1，並tryTerminate
        if (! workerStarted)
            addWorkerFailed(w);
    }
    return workerStarted;
}

// 執行中斷的方法
private void interruptWorkers() {
   final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        for (Worker w : workers)
            w.interruptIfStarted();
    } finally {
        mainLock.unlock();
    }
}

void interruptIfStarted() {
  Thread t;
  	// 只有state >= 0 的線程才能夠執行中斷操做，因此構造的worker state = -1
    if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
        try {
            t.interrupt();
        } catch (SecurityException ignore) {
        }
    }
}
複製代碼

核心方法3：run()

// 執行任務內部調用runWorker(this)
public void run() {
    runWorker(this);
}

// 執行runWorker(Worker w)
final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    // 
    w.unlock();
    // 控制processWorkerExit是否處理workerCount，默認true
    boolean completedAbruptly = true;
    try {
        while (task != null || (task = getTask()) != null) {
            w.lock();
            // 若是池正在中止，確保線程被中斷；
            // 若是沒有，確保線程不被中斷。 
            // 這須要在第二種狀況下從新檢查以處理shutdownNow，同時清除中斷
            if ((runStateAtLeast(ctl.get(), STOP) ||
                 (Thread.interrupted() &&
                  runStateAtLeast(ctl.get(), STOP))) &&
                !wt.isInterrupted())
                wt.interrupt();
            try {
                // 執行前置方法（見示例：可繼承重寫）
                beforeExecute(wt, task);
                Throwable thrown = null;
                try {
                    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 = null;
                w.completedTasks++;
                w.unlock();
            }
        }
        completedAbruptly = false;
    } finally {
        processWorkerExit(w, completedAbruptly);
    }
}

// worker退出處理
private void processWorkerExit(Worker w, boolean completedAbruptly) {
	// 若是是異常中斷則不調整workerCount
    if (completedAbruptly)
        decrementWorkerCount();

	// 獲取獨佔鎖
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
    	// 設置線程池完成任務數 += 當前worker完成任務數
        completedTaskCount += w.completedTasks;
        // 從worker隊列中移除當前worker
        workers.remove(w);
    } finally {
        mainLock.unlock();
    }

	// 嘗試設置線程池狀態爲TERMINATED
    tryTerminate();

    int c = ctl.get();
    if (runStateLessThan(c, STOP)) {
        if (!completedAbruptly) {
            int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
            if (min == 0 && ! workQueue.isEmpty())
                min = 1;
            if (workerCountOf(c) >= min)
                return; // replacement not needed
        }
        addWorker(null, false);
    }
}
複製代碼

核心方法4：shutDown()和shutDownNow()

public void shutdown() {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
		// 檢查當前線程是否有shutDown權限
        checkShutdownAccess();
		// 設置線程池狀態，若是已是該狀態直接返回
        advanceRunState(SHUTDOWN);
		// 若是工做線程沒有中斷，而且沒有正在運行則設置中斷標誌
        interruptIdleWorkers();
        onShutdown(); // hook for ScheduledThreadPoolExecutor
    } finally {
        mainLock.unlock();
    }
	// 嘗試設置線程池狀態爲TERMINATED
    tryTerminate();
}

public List<Runnable> shutdownNow() {
    List<Runnable> tasks;
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        // 檢查當前線程是否有shutDown權限
        checkShutdownAccess();
		// 設置線程池狀態，若是已是該狀態直接返回
        advanceRunState(STOP);
		// 中斷全部工做線程
        interruptWorkers();
		// 將隊列中的元素移動到tasks列表
        tasks = drainQueue();
    } finally {
        mainLock.unlock();
    }
	// 嘗試設置線程池狀態爲TERMINATED
    tryTerminate();
	// 返回隊列中被丟棄的任務列表
    return tasks;
}
複製代碼

核心方法5：awaitTermination(long timeout, TimeUnit unit)

該方法調用會被阻塞，如下幾種狀況任意一個發生了就會致使該方法的執行：this

全部任務執行完畢而且shutdown請求被調用
參數中定義的timeout時間到達
當前線程被打斷

核心方法6：tryTerminate()

/** * 1.線程池處於RUNNING狀態 * 2.線程池已經處於TERMINATE * 3.線程池爲SHUTDOWN狀態而且隊列不爲空 * 以上三種狀況直接return */
final void tryTerminate() {
    for (;;) {
        int c = ctl.get();
        if (isRunning(c) ||
            runStateAtLeast(c, TIDYING) ||
            (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
            return;
        // 有資格終止
        if (workerCountOf(c) != 0) {
            interruptIdleWorkers(ONLY_ONE);
            return;
        }
		// 獲取鎖嘗試終止
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
        	// CAS修改線程池狀態爲TIDYING
            if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
                try {
                	// 成功則執行terminated方法（見示例：可繼承重寫）
                    terminated();
                } finally {
                	// 設置線程池狀態爲TERMINATED
                    ctl.set(ctlOf(TERMINATED, 0));
                    // 喚醒termination條件隊列中的線程（核心方法5 awaitTermination）
                    termination.signalAll();
                }
                return;
            }
        } finally {
            mainLock.unlock();
        }
        // else retry on failed CAS
    }
}
複製代碼

示例：繼承ThreadPoolExecutor重寫部分方法

public class ThreadPoolExecutorTest extends ThreadPoolExecutor{

    public ThreadPoolExecutorTest( int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue ) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
    }

    @Override
    protected void terminated() {
        System.out.println("===它終止啦它終止啦！===");
        super.terminated();
    }

    @Override
    public void execute( Runnable command ) {
        System.out.println("有人execute我");
        super.execute(command);
    }

    @Override
    protected void beforeExecute( Thread t, Runnable r ) {
        System.out.println("===它快來啦它快來啦！===");
        super.beforeExecute(t, r);
    }

    @Override
    protected void afterExecute( Runnable r, Throwable t ) {
        System.out.println("===它走啦它走啦！===");
        super.afterExecute(r, t);
    }

    @SneakyThrows
    public static void main( String[] args ) {
        ThreadPoolExecutorTest threadPool = new ThreadPoolExecutorTest(1, 1, 0, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>());
        threadPool.execute(new Thread(() -> System.out.println("幹活！！！")));
        threadPool.awaitTermination(5, TimeUnit.SECONDS);
        System.out.println("wait 超時");
        threadPool.shutdownNow();
        threadPool.execute(new Thread(() -> System.out.println("我又來幹活兒啦！")));
    }

}

// output shutdown以後的線程不能再次execute激活
有人execute我
===它快來啦它快來啦！===
幹活！！！
===它走啦它走啦！===
wait 超時
===它終止啦它終止啦！===
有人execute我
Exception in thread "main" java.util.concurrent.RejectedExecutionException: Task Thread[Thread-1,5,main] rejected from *.*.*.juc.ThreadPoolExecutorTest@6193b845[Terminated, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 1]
	at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2063)
	at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:830)
	at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1379)
	at *.*.*.juc.ThreadPoolExecutorTest.execute(ThreadPoolExecutorTest.java:32)
	at *.*.*.juc.ThreadPoolExecutorTest.main(ThreadPoolExecutorTest.java:55)
複製代碼

總結

線程池巧妙地使用了AtomicInteger來記錄線程池的狀態（高3位）和線程池中的線程個數（低29位）。經過線程池狀態來控制任務的執行，每一個Worker線程能夠處理多個任務。線程池經過線程的複用減小了線程建立和銷燬的開銷。spa

相關標籤/搜索

每日一句

每一个你不满意的现在，都有一个你没有努力的曾经。