//AbstractExecutorService.java文件
? ? // executorService 中的 submit 方法
? ? public Future<?> submit(Runnable task) {
? ? ? ? // 首先判斷傳入的runnable 對(duì)象是否為空
? ? ? ? if (task == null) throw new NullPointerException();
? ? ? ? // 創(chuàng)建一個(gè) futuretask 對(duì)象
? ? ? ? RunnableFuture<Void> ftask = newTaskFor(task, null);
? ? ? ? execute(ftask);
? ? ? ? return ftask;
? ? }
? ? // 根據(jù)runnable 創(chuàng)建一個(gè)futuretask對(duì)象
? ? protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
? ? ? ? return new FutureTask<T>(runnable, value);
? ? }
? ? // ThreadPoolExecutor.java文件
? ? // 執(zhí)行創(chuàng)建線程池的方法
? ? public void execute(Runnable command) {
? ? ? ? // 首先判斷傳入的線程是否為空
? ? ? ? if (command == null)
? ? ? ? ? ? // 為空,拋出異常
? ? ? ? ? ? throw new NullPointerException();
? ? ? ? // 獲取線程池的狀態(tài)碼, 這個(gè)狀態(tài)碼是自增的,原子類(lèi)型的自增, 在執(zhí)行addworker后ctl會(huì)加1
? ? ? ? int c = ctl.get();
? ? ? ? // 通過(guò)狀態(tài)碼,獲取線程池中的線程的數(shù)量,如果小于核心數(shù)量
? ? ? ? if (workerCountOf(c) < corePoolSize) {
? ? ? ? ? ? // 添加線程到線程池,并且為true時(shí)使用核心線程數(shù)作為邊界,如果false ,使用最大數(shù)量線程數(shù)作為邊界
? ? ? ? ? ? if (addWorker(command, true))
? ? ? ? ? ? ? ? // 添加完成后,返回
? ? ? ? ? ? ? ? return;
? ? ? ? ? ? // 如果添加失敗,重新獲取狀態(tài)值
? ? ? ? ? ? c = ctl.get();
? ? ? ? }
? ? ? ? // 執(zhí)行下面邏輯有兩種情況
? ? ? ? // ? ? ?1. 工作線程數(shù)大于核心線程
? ? ? ? // ? ? ?2. 添加線程時(shí)出錯(cuò)
? ? ? ? // 如果線程池中線程的數(shù)量大于核心的數(shù)量, 判斷如果是運(yùn)行狀態(tài), 并且也把線程加進(jìn)了阻塞隊(duì)列 workQueue 中
? ? ? ? if (isRunning(c) && workQueue.offer(command)) {
? ? ? ? ? ? // 重新獲取 線程池 狀態(tài)值
? ? ? ? ? ? int recheck = ctl.get();
? ? ? ? ? ? // 判斷當(dāng)前線程池如果不是運(yùn)行狀態(tài),并且成功從隊(duì)列中移除(從workQueue中移除線程, 并嘗試終止線程池)
? ? ? ? ? ? if (! isRunning(recheck) && remove(command))
? ? ? ? ? ? ? ? // 執(zhí)行拒絕執(zhí)行線程的處理
? ? ? ? ? ? ? ? reject(command);
? ? ? ? ? ? ? ? // 如果工作線程數(shù)為0
? ? ? ? ? ? else if (workerCountOf(recheck) == 0)
? ? ? ? ? ? ? ? // 添加一個(gè)null的工作包裝對(duì)象
? ? ? ? ? ? ? ? addWorker(null, false);
? ? ? ? } else if (!addWorker(command, false))
? ? ? ? ? ? // 如果添加到線程池中出錯(cuò),執(zhí)行拒接的線程
? ? ? ? ? ? reject(command);
? ? }
? ? // 創(chuàng)建一個(gè)原子類(lèi)對(duì)象用于計(jì)算線程的中狀態(tài)
? ? private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
? ? // integer.size 為 32
? ? private static final int COUNT_BITS = Integer.SIZE - 3;
? ? private static final int CAPACITY ? = (1 << COUNT_BITS) - 1;
? ? // 即高3位為1，低29位為0，該狀態(tài)的線程池會(huì)接收新任務(wù)，也會(huì)處理在阻塞隊(duì)列中等待處理的任務(wù)
? ? private static final int RUNNING ? ?= -1 << COUNT_BITS;
? ? // 即高3位為0，低29位為0，該狀態(tài)的線程池不會(huì)再接收新任務(wù)，但還會(huì)處理已經(jīng)提交到阻塞隊(duì)列中等待處理的任務(wù)
? ? private static final int SHUTDOWN ? = ?0 << COUNT_BITS;
? ? // 即高3位為001，低29位為0，該狀態(tài)的線程池不會(huì)再接收新任務(wù)，不會(huì)處理在阻塞隊(duì)列中等待的任務(wù)，而且還會(huì)中斷正在運(yùn)行的任務(wù)
? ? private static final int STOP ? ? ? = ?1 << COUNT_BITS;
? ? // 即高3位為010，低29位為0，所有任務(wù)都被終止了，workerCount為0，為此狀態(tài)時(shí)還將調(diào)用terminated()方法
? ? private static final int TIDYING ? ?= ?2 << COUNT_BITS;
? ? // 即高3位為100，低29位為0，terminated()方法調(diào)用完成后變成此狀態(tài) ?
? ? private static final int TERMINATED = ?3 << COUNT_BITS;
? ? // 用戶(hù)計(jì)算線程的狀態(tài) 32位中 高3位為1 低29位為0?
? ? private static int runStateOf(int c) ? ? { return c & ~CAPACITY; }
? ? // 用于計(jì)算線程池中線程的數(shù)量 32位中 高3位為0 ?低29位為1
? ? private static int workerCountOf(int c) ?{ return c & CAPACITY; }
? ? // rs 為 runState, wc 為 workerCount 通過(guò)工作狀態(tài)和線程數(shù)量來(lái)計(jì)算出 ctl
? ? private static int ctlOf(int rs, int wc) { return rs | wc; }
? ? // 添加工作線程的方法
? ? private boolean addWorker(Runnable firstTask, boolean core) {
? ? ? ? // 設(shè)置循環(huán)跳出點(diǎn),如果執(zhí)行到某個(gè)位置,使用break,直接跳出的是這個(gè)標(biāo)簽范圍內(nèi)的所有循環(huán)
? ? ? ? retry:
? ? ? ? for (;;) {
? ? ? ? ? ? // 獲取線程狀態(tài)
? ? ? ? ? ? int c = ctl.get();
? ? ? ? ? ? int rs = runStateOf(c);
? ? ? ? ? ? // 判斷線程池狀態(tài)是否在shutdown上以及 狀態(tài)不是關(guān)閉并且添加的線程不為空,并且線程隊(duì)列中的線程不是空的
? ? ? ? ? ? if (rs >= SHUTDOWN && !(rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty()))
? ? ? ? ? ? ? ? // 如果滿(mǎn)足上面條件,說(shuō)明線程池已經(jīng)不適合添加新的線程了, 直接返回false
? ? ? ? ? ? ? ? return false;
? ? ? ? ? ? // 如果不滿(mǎn)足上面條件,說(shuō)明線程池可以添加線程, 下面這個(gè)循環(huán)主要是對(duì)ctl進(jìn)行操作,保證在增1后線程狀態(tài)保持同步
? ? ? ? ? ? for (;;) {
? ? ? ? ? ? ? ? // 獲取工作線程數(shù)量
? ? ? ? ? ? ? ? int wc = workerCountOf(c);
? ? ? ? ? ? ? ? // 判斷當(dāng)前線程池中工作線程數(shù)量是否大于線程容量,大于核心線程數(shù)或最大線程數(shù)
? ? ? ? ? ? ? ? if (wc >= CAPACITY ||
? ? ? ? ? ? ? ? ? ? wc >= (core ? corePoolSize : maximumPoolSize))
? ? ? ? ? ? ? ? ? ? // 滿(mǎn)足條件,說(shuō)明當(dāng)前線程不是適合添加新的線程的
? ? ? ? ? ? ? ? ? ? return false;
? ? ? ? ? ? ? ? // 如果工作數(shù)量少于最大量或者核心線程數(shù)或最大線程數(shù), 工作線程數(shù)加1,即操作ctl,通過(guò)cas的方式
? ? ? ? ? ? ? ? if (compareAndIncrementWorkerCount(c))
? ? ? ? ? ? ? ? ? ? // 如果添加成功,跳出內(nèi)循環(huán),
? ? ? ? ? ? ? ? ? ? break retry;
? ? ? ? ? ? ? ? // 如果添加失敗,重新獲取ctl
? ? ? ? ? ? ? ? c = ctl.get(); ?// Re-read ctl
? ? ? ? ? ? ? ? // 判斷此時(shí)線程池狀態(tài)是否已經(jīng)改變
? ? ? ? ? ? ? ? if (runStateOf(c) != rs)
? ? ? ? ? ? ? ? ? ? //如果狀態(tài)不一致,跳過(guò),重新循環(huán)
? ? ? ? ? ? ? ? ? ? continue retry;
? ? ? ? ? ? ? ? // else CAS failed due to workerCount change; retry inner loop
? ? ? ? ? ? }
? ? ? ? }
? ? ? ? boolean workerStarted = false;
? ? ? ? boolean workerAdded = false;
? ? ? ? // 創(chuàng)建一個(gè)線程包裝對(duì)象,用于包裝線程
? ? ? ? Worker w = null;
? ? ? ? try {
? ? ? ? ? ? w = new Worker(firstTask);
? ? ? ? ? ? // 創(chuàng)建一個(gè)worker 工作線程
? ? ? ? ? ? final Thread t = w.thread;
? ? ? ? ? ? // 判斷創(chuàng)建的線程是否為空
? ? ? ? ? ? if (t != null) {
? ? ? ? ? ? ? ? ?// 如果不為空,獲取鎖對(duì)象
? ? ? ? ? ? ? ? final ReentrantLock mainLock = this.mainLock;
? ? ? ? ? ? ? ? // 開(kāi)始加鎖
? ? ? ? ? ? ? ? mainLock.lock();
? ? ? ? ? ? ? ? try {
? ? ? ? ? ? ? ? ? ? // 獲取線程池狀態(tài)
? ? ? ? ? ? ? ? ? ? int rs = runStateOf(ctl.get());
? ? ? ? ? ? ? ? ? ? // 如果線程池狀態(tài)是running或者線程池狀態(tài)關(guān)閉并且傳入的線程是空的
? ? ? ? ? ? ? ? ? ? if (rs < SHUTDOWN ||
? ? ? ? ? ? ? ? ? ? ? ? (rs == SHUTDOWN && firstTask == null)) {
? ? ? ? ? ? ? ? ? ? ? ? // 判斷創(chuàng)建的工作線程是否是活動(dòng)狀態(tài)(已經(jīng)開(kāi)始還沒(méi)有死掉)
? ? ? ? ? ? ? ? ? ? ? ? if (t.isAlive()) // precheck that t is startable
? ? ? ? ? ? ? ? ? ? ? ? ? ? // 如果是活動(dòng)狀態(tài),拋出 非法線程狀態(tài)異常?
? ? ? ? ? ? ? ? ? ? ? ? ? ? throw new IllegalThreadStateException();
? ? ? ? ? ? ? ? ? ? ? ? // 如果不是活動(dòng)狀態(tài), 添加到set集合中,這個(gè)set集合只有持有mainlock才可以訪問(wèn)
? ? ? ? ? ? ? ? ? ? ? ? workers.add(w);
? ? ? ? ? ? ? ? ? ? ? ? // 獲取集合長(zhǎng)度
? ? ? ? ? ? ? ? ? ? ? ? int s = workers.size();
? ? ? ? ? ? ? ? ? ? ? ? // 如果存放剛才創(chuàng)建的workers工作線程的集合中的線程數(shù)超過(guò)最大的池的大小
? ? ? ? ? ? ? ? ? ? ? ? if (s > largestPoolSize)
? ? ? ? ? ? ? ? ? ? ? ? ? ? // 把set集合中的數(shù)量代替原線程池最大值
? ? ? ? ? ? ? ? ? ? ? ? ? ? largestPoolSize = s;
? ? ? ? ? ? ? ? ? ? ? ? workerAdded = true;
? ? ? ? ? ? ? ? ? ? }
? ? ? ? ? ? ? ? } finally {
? ? ? ? ? ? ? ? ? ? // 釋放鎖
? ? ? ? ? ? ? ? ? ? mainLock.unlock();
? ? ? ? ? ? ? ? }
? ? ? ? ? ? ? ? // 根據(jù)前面的判斷是否需要開(kāi)啟線程,如果線程已經(jīng)是活動(dòng)的,不需要開(kāi)啟,如果不是活動(dòng)線程,開(kāi)啟線程
? ? ? ? ? ? ? ? if (workerAdded) {
? ? ? ? ? ? ? ? ? ? t.start();
? ? ? ? ? ? ? ? ? ? // 開(kāi)啟成功,設(shè)置workerStarted 為 true
? ? ? ? ? ? ? ? ? ? workerStarted = true;
? ? ? ? ? ? ? ? }
? ? ? ? ? ? }
? ? ? ? } finally {
? ? ? ? ? ? // 如果工作線程開(kāi)啟失敗,調(diào)用添加到失敗的線程中
? ? ? ? ? ? if (! workerStarted)
? ? ? ? ? ? ? ? // 從set中移除失敗的線程,并且ctl減1, 并且嘗試終止線程池
? ? ? ? ? ? ? ? addWorkerFailed(w);
? ? ? ? }
? ? ? ? return workerStarted;
? ? }
? ? // 線程開(kāi)啟失敗后的方法
? ? private void addWorkerFailed(Worker w) {
? ? ? ? // 獲取鎖
? ? ? ? final ReentrantLock mainLock = this.mainLock;
? ? ? ? mainLock.lock();
? ? ? ? try {
? ? ? ? ? ? if (w != null)
? ? ? ? ? ? ? ? // 如果線程不為空,從set集合中移除沒(méi)有開(kāi)啟成功的線程
? ? ? ? ? ? ? ? workers.remove(w);
? ? ? ? ? ? // 減去之前ctl增加的1
? ? ? ? ? ? decrementWorkerCount();
? ? ? ? ? ? // 嘗試中斷線程
? ? ? ? ? ? tryTerminate();
? ? ? ? } finally {
? ? ? ? ? ? mainLock.unlock();
? ? ? ? }
? ? }
? ? // 通過(guò)cas方式ctl加1
? ? private boolean compareAndIncrementWorkerCount(int expect) {
? ? ? ? return ctl.compareAndSet(expect, expect + 1);
? ? }
? ? // 移除線程
? ? public boolean remove(Runnable task) {
? ? ? ? // 從等待隊(duì)列中一尺線程
? ? ? ? boolean removed = workQueue.remove(task);
? ? ? ? // 嘗試終止線程池
? ? ? ? tryTerminate(); // In case SHUTDOWN and now empty
? ? ? ? return removed;
? ? }
? ? // 使用拒絕處理對(duì)象執(zhí)行拒接指定線程
? ? final void reject(Runnable command) {
? ? ? ? handler.rejectedExecution(command, this);
? ? }