Executor --> ExecutorService --> AbstractExecutorService --> ThreadPoolExecuotorExecutor接口，只有一個execute方法，參數為線程任務ExecutorService接口繼承Executor接口，增加了submit、shutdown、invokeAll等方法AbstractExecutorService抽象類，實現ExecutorService接口，提供了submit、invokeAny、invokeAll默認實現方法， execute、 shutdown、shutdownNow等待沒有提供默認的實現ThreadPoolExecutor線程池狀態值---RUNNING 接受新任務并且處理阻塞隊列里的任務---SHUTDOWN 拒絕信任務但是處理阻塞隊列里的任務---STOP 拒絕新任務并且拋棄阻塞隊列里的任務同時會中斷正在處理的任務---TIDYING 所有任務都執行完當前線程池活動線程為0，將調用terminated方法---TERMINATED 終止狀態構造函數---有四個構造函數，其他三個都是調用下面這個構造函數public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) {}---參數說明corePoolSize 核心線程數maximumPoolSize 最大線程數keepAliveTime 存活時間unit 時間單位workQueue 存放線程的隊列threadFactory 創建線程的工廠handler 拒絕策略提交任務---submitpublic Future<?> submit(Runable task) {if(task == null) throw new NullPointerException();RunnableFuture<Void> ftask = newTaskFor(task, null);execute(ftask);return ftask;} public <T> Future<T> submit(Runnable task, T result) {if(task == null) throw new NullPointerExecetion();RunnableFuture<T> ftask = newTaskFor(task, result);execute(ftask);return ftask;}public <T> Future<T> submit(Callable<T> task) {if(task == null) throw new NullPointerException();RunnableFuture<T> ftask = newTaskFor(task);execute(ftask);return ftask;}--- executepublic void execute(Runnable command) {if(command == null) throw new NullPointerException();//獲取當前線程池的狀態+線程個數int c = ctl.get();//判斷當前線程池數量是否小于corePoolSize，小于則調用addWorker方法創建新線程運行，且傳進來的Runnable當做第一個任務執行if(workerCountOf(c) < corePoolSize) {if(addWorker(command, true))return;c = ctl.get(); }//如果線程池處于Running狀態，則添加任務到阻塞隊列if(isRunning(c) && workQueue.offer(command)) {int recheck = ctl.get();//如果當前線程池狀態不是Running則從隊列刪除任務，并執行拒絕策略if(!isRunning(recheck) && remove(command))reject(command);//如果當前線程池為空，則添加一個線程 else if(workercountOf(recheck) == 0) addWorker(null, false); }//新增線程失敗則執行拒絕策略else if(!addWorker(command, false))reject(command);}---addWorkerprivate boolean addWorder(Runnable firstTask, boolean core) {retry:for(;;) {int c = ctl.get();int rs = runStateOf(c);//檢查當前線程池狀態是否是shutdown、stop、tidying、terminated且!(當前狀態為shutdown、傳入的任務為null、隊列不為null),條件都成立時返回falseif (rs >= SHUTDOWN &&! (rs == SHUTDOWN &&firstTask == null &&! workQueue.isEmpty()))return false;for(;;) {int wc = workerCountOf(c);if(wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize) )return false;if(compareAndIncrementWorkerCount(c))break retry;c = ctl.get();if(runStateOf(c) != rs) continue retry; }}boolean workerStarted = false;boolean workerAdded = false;Worker w = null;try{// new threadw = new Worker(firstTask);final Thread t = w.thread();if(t != null) {//lockfinal ReentrantLock mainLock = this.mainLock;mainLock.lock();try{//check thread pool stateint rs = runStateOf(ctl.get());if(rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null) ) {//檢查線程是否可啟動if(t.isAlive())throw new IllegalThreadStateException();workers.add(w) ;int s = workers.size();if(s > largestPoolSize)largestPoolSize = s;workerAdded = true; } }finally {mainLock.unlock();}//判斷worker是否添加成功，成功則啟動線程，將workerStarted設置為trueif(workerAdded) {t.start();workerStarted = true;}}}finally{if(!workerStarted) addWorkerFailed(w);}return workerStarted;}---runWorker,線程啟動時調用了runWorker方法final void runWorker(Worker w) {Thread wt = Thread.currentThread();Runnable task = w.firstTask;w.firstTask = null;w.unlock();boolean completedAbruptly = true;try {//循環獲取任務while(task != null || (task = getTask()) != null ) {w.lock();// 當線程池是處于STOP狀態或者TIDYING、TERMINATED狀態時，設置當前線程處于中斷狀態// 如果不是，當前線程就處于RUNNING或者SHUTDOWN狀態，確保當前線程不處于中斷狀態// 重新檢查當前線程池的狀態是否大于等于STOP狀態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 Error(x);}finally {afterExecute(taskm thrown);}}finally {task = null;w.completedTasks++;w.unlock();} }completedAbruptly = false;}finally {processWokerExit(w, completedAbruptly);}}---getTaskprivate Runnable getTask() {boolean timeOut = false;for(;;) {int c = ctl.get();int rs = runStateOf(c);if(rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty() )) {decrementWorkerCount();return null;}int wc = workerCountof(c);boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;//（當前線程數是否大于最大線程數或者）//且（線程數大于1或者任務隊列為空）//這里有個問題(timed && timedOut)timedOut = false，好像(timed && timedOut)一直都是false吧if ((wc > maximumPoolSize || (timed && timedOut))&& (wc > 1 || workQueue.isEmpty())) {if (compareAndDecrementWorkerCount(c))return null;continue;}try{Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take();if(r != null)return r;timedOut = true; }catch(InterruptedException retry) {timedOut = false;}}}---shutdown 關閉線程池public void shutdown() {final ReentrantLock mainLock = this.mainLock;mainLock.lock();try{checkShutdownAccess();advanceRunState(SHUTDOWN);interruptIdleWorkers();onShutdown();}finally{mainLock.unlock();}tryTerminate();}

?

轉載于:https://www.cnblogs.com/codechange/p/9254059.html

總結

以上是生活随笔為你收集整理的ThreadPoolExecuotor源码参考的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。