一、總覽

線程池類ThreadPoolExecutor的相關(guān)類需要先了解：

?（圖片來自：https://javadoop.com/post/java-thread-pool#%E6%80%BB%E8%A7%88）

Executor：位于最頂層，只有一個 execute(Runnable runnable) 方法，用于提交任務(wù)。

ExecutorService ：在 Executor 接口的基礎(chǔ)上添加了很多的接口方法，提交任務(wù)，獲取結(jié)果，關(guān)閉線程池。

AbstractExecutorService：實(shí)現(xiàn)了ExecutorService 接口，然后在其基礎(chǔ)上實(shí)現(xiàn)了幾個實(shí)用的方法，這些方法提供給子類進(jìn)行調(diào)用。

ThreadPoolExecutor：線程池類

Executors：最常用的用于生成 ThreadPoolExecutor 的實(shí)例的工具類

FutureTask：Runnable, Future -> RunnableFuture -> FutureTask

FutureTask 通過 RunnableFuture 間接實(shí)現(xiàn)了 Runnable 接口， 所以每個 Runnable 通常都先包裝成 FutureTask， 然后調(diào)用 executor.execute(Runnable command) 將其提交給線程池

Runnable 的 void run() 方法是沒有返回值的，如果我們需要的話，會在 submit 中指定第二個參數(shù)作為返回值。

Callable：Callable 也是因為線程池的需要，所以才有了這個接口。它和 Runnable 的區(qū)別在于 run() 沒有返回值，而 Callable 的 call() 方法有返回值

BlockingQueue：Java并發(fā)（十八）：阻塞隊列BlockingQueue

二、線程池狀態(tài)

線程池中的各個狀態(tài)：

• RUNNING：這個沒什么好說的，這是最正常的狀態(tài)：接受新的任務(wù)，處理等待隊列中的任務(wù)
• SHUTDOWN：不接受新的任務(wù)提交，但是會繼續(xù)處理等待隊列中的任務(wù)
• STOP：不接受新的任務(wù)提交，不再處理等待隊列中的任務(wù)，中斷正在執(zhí)行任務(wù)的線程
• TIDYING：所有的任務(wù)都銷毀了，workCount 為 0。線程池的狀態(tài)在轉(zhuǎn)換為 TIDYING 狀態(tài)時，會執(zhí)行鉤子方法 terminated()
• TERMINATED：terminated() 方法結(jié)束后，線程池的狀態(tài)就會變成這個

狀態(tài)轉(zhuǎn)換：

• RUNNING -> SHUTDOWN：當(dāng)調(diào)用了 shutdown() 后，會發(fā)生這個狀態(tài)轉(zhuǎn)換，這也是最重要的
• (RUNNING or SHUTDOWN) -> STOP：當(dāng)調(diào)用 shutdownNow() 后，會發(fā)生這個狀態(tài)轉(zhuǎn)換，這下要清楚 shutDown() 和 shutDownNow() 的區(qū)別了
• SHUTDOWN -> TIDYING：當(dāng)任務(wù)隊列和線程池都清空后，會由 SHUTDOWN 轉(zhuǎn)換為 TIDYING
• STOP -> TIDYING：當(dāng)任務(wù)隊列清空后，發(fā)生這個轉(zhuǎn)換
• TIDYING -> TERMINATED：這個前面說了，當(dāng) terminated() 方法結(jié)束后

ThreadPoolExecutor采用一個 32 位的整數(shù)來存放線程池的狀態(tài)和當(dāng)前池中的線程數(shù)，其中高 3 位用于存放線程池狀態(tài)，低 29 位表示線程數(shù)。

private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));// 這里 COUNT_BITS 設(shè)置為 29(32-3)，意味著前三位用于存放線程狀態(tài)，后29位用于存放線程數(shù)// 很多初學(xué)者很喜歡在自己的代碼中寫很多 29 這種數(shù)字，或者某個特殊的字符串，然后分布在各個地方，這是非常糟糕的private static final int COUNT_BITS = Integer.SIZE - 3;// 000 11111111111111111111111111111// 這里得到的是 29 個 1，也就是說線程池的最大線程數(shù)是 2^29-1=536870911// 以我們現(xiàn)在計算機(jī)的實(shí)際情況，這個數(shù)量還是夠用的private static final int CAPACITY = (1 << COUNT_BITS) - 1;// 我們說了，線程池的狀態(tài)存放在高 3 位中// 運(yùn)算結(jié)果為 111跟29個0：111 00000000000000000000000000000private static final int RUNNING = -1 << COUNT_BITS;// 000 00000000000000000000000000000private static final int SHUTDOWN = 0 << COUNT_BITS;// 001 00000000000000000000000000000private static final int STOP = 1 << COUNT_BITS;// 010 00000000000000000000000000000private static final int TIDYING = 2 << COUNT_BITS;// 011 00000000000000000000000000000private static final int TERMINATED = 3 << COUNT_BITS;// 將整數(shù) c 的低 29 位修改為 0，就得到了線程池的狀態(tài)private static int runStateOf(int c) { return c & ~CAPACITY; }// 將整數(shù) c 的高 3 為修改為 0，就得到了線程池中的線程數(shù)private static int workerCountOf(int c) { return c & CAPACITY; }private static int ctlOf(int rs, int wc) { return rs | wc; }private static boolean runStateLessThan(int c, int s) {return c < s;}private static boolean runStateAtLeast(int c, int s) {return c >= s;}private static boolean isRunning(int c) {return c < SHUTDOWN;}

三、線程池參數(shù)

通過ThreadPoolExecutor構(gòu)造函數(shù)來看線程池參數(shù)：

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

corePoolSize：線程池中核心線程的數(shù)量。當(dāng)提交一個任務(wù)時，線程池會新建一個線程來執(zhí)行任務(wù)，直到當(dāng)前線程數(shù)等于corePoolSize。如果調(diào)用了線程池的prestartAllCoreThreads()方法，線程池會提前創(chuàng)建并啟動所有基本線程。

maximumPoolSize：線程池中允許的最大線程數(shù)。線程池的阻塞隊列滿了之后，如果還有任務(wù)提交，如果當(dāng)前的線程數(shù)小于maximumPoolSize，則會新建線程來執(zhí)行任務(wù)。注意，如果使用的是無界隊列，該參數(shù)也就沒有什么效果了。

keepAliveTime：空閑線程的保活時間，如果某線程的空閑時間超過這個值都沒有任務(wù)給它做，那么可以被關(guān)閉了。注意這個值并不會對所有線程起作用，如果線程池中的線程數(shù)少于等于核心線程數(shù) corePoolSize，那么這些線程不會因為空閑太長時間而被關(guān)閉，當(dāng)然，也可以通過調(diào)用?allowCoreThreadTimeOut(true)使核心線程數(shù)內(nèi)的線程也可以被回收

unit：keepAliveTime的單位。TimeUnit

workQueue：

用來保存等待執(zhí)行的任務(wù)的阻塞隊列，等待的任務(wù)必須實(shí)現(xiàn)Runnable接口。我們可以選擇如下幾種：

• ArrayBlockingQueue：基于數(shù)組結(jié)構(gòu)的有界阻塞隊列，FIFO。
  
• LinkedBlockingQueue：基于鏈表結(jié)構(gòu)的有界阻塞隊列，FIFO。
• SynchronousQueue：不存儲元素的阻塞隊列，每個插入操作都必須等待一個移出操作，反之亦然。

threadFactory：用于設(shè)置創(chuàng)建線程的工廠。

handler：

RejectedExecutionHandler，線程池的拒絕策略。

所謂拒絕策略，是指將任務(wù)添加到線程池中時，線程池拒絕該任務(wù)所采取的相應(yīng)策略。當(dāng)向線程池中提交任務(wù)時，如果此時線程池中的線程已經(jīng)飽和了，而且阻塞隊列也已經(jīng)滿了，則線程池會選擇一種拒絕策略來處理該任務(wù)。

線程池提供了四種拒絕策略：（重寫RejectedExecutionHandler.rejectedExecution(Runnable, ThreadPoolExecutor)）

　　AbortPolicy：直接拋出異常，默認(rèn)策略；

　　CallerRunsPolicy：用調(diào)用者所在的線程來執(zhí)行任務(wù)；

　　DiscardOldestPolicy：丟棄阻塞隊列中靠最前的任務(wù)，并執(zhí)行當(dāng)前任務(wù)；

　　DiscardPolicy：直接丟棄任務(wù)；?

當(dāng)然我們也可以實(shí)現(xiàn)自己的拒絕策略，例如記錄日志等等，實(shí)現(xiàn)RejectedExecutionHandler接口寫rejectedExecution方法即可。

四、線程池創(chuàng)建

Executor工具類提供了三種線程池創(chuàng)建方式：

FixedThreadPool ：可重用固定線程數(shù)的線程池

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

corePoolSize 和 maximumPoolSize都設(shè)置為創(chuàng)建FixedThreadPool時指定的參數(shù)nThreads，意味著當(dāng)線程池滿時且阻塞隊列也已經(jīng)滿時，如果繼續(xù)提交任務(wù)，則會直接走拒絕策略，該線程池不會再新建線程來執(zhí)行任務(wù)，而是直接走拒絕策略。FixedThreadPool使用的是默認(rèn)的拒絕策略，即AbortPolicy，則直接拋出異常。

keepAliveTime設(shè)置為0L，表示空閑的線程會立刻終止。

workQueue則是使用LinkedBlockingQueue，但是沒有設(shè)置范圍，那么則是最大值（Integer.MAX_VALUE），這基本就相當(dāng)于一個無界隊列了。使用該“無界隊列”則會帶來哪些影響呢？當(dāng)線程池中的線程數(shù)量等于corePoolSize 時，如果繼續(xù)提交任務(wù)，該任務(wù)會被添加到阻塞隊列workQueue中，當(dāng)阻塞隊列也滿了之后，則線程池會新建線程執(zhí)行任務(wù)直到maximumPoolSize。由于FixedThreadPool使用的是“無界隊列”LinkedBlockingQueue，那么maximumPoolSize參數(shù)無效，同時指定的拒絕策略AbortPolicy也將無效。而且該線程池也不會拒絕提交的任務(wù)，如果客戶端提交任務(wù)的速度快于任務(wù)的執(zhí)行，那么keepAliveTime也是一個無效參數(shù)。

SingleThreadExecutor：只有一個線程的固定線程池

public static ExecutorService newSingleThreadExecutor() {return new FinalizableDelegatedExecutorService(new ThreadPoolExecutor(1, 1,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>()));}

為單一worker線程的線程池，SingleThreadExecutor把corePool和maximumPoolSize均被設(shè)置為1，和FixedThreadPool一樣使用的是無界隊列LinkedBlockingQueue,所以帶來的影響和FixedThreadPool一樣。

CachedThreadPool：根據(jù)需要創(chuàng)建新線程的線程池

public static ExecutorService newCachedThreadPool() {return new ThreadPoolExecutor(0, Integer.MAX_VALUE,60L, TimeUnit.SECONDS,new SynchronousQueue<Runnable>());}

CachedThreadPool的corePool為0，maximumPoolSize為Integer.MAX_VALUE，這就意味著所有的任務(wù)一提交就會加入到阻塞隊列中。

keepAliveTime這是為60L，unit設(shè)置為TimeUnit.SECONDS，意味著空閑線程等待新任務(wù)的最長時間為60秒，空閑線程超過60秒后將會被終止。

阻塞隊列采用的SynchronousQueue，SynchronousQueue是一個沒有元素的阻塞隊列，加上corePool = 0 ，maximumPoolSize = Integer.MAX_VALUE，這樣就會存在一個問題，如果主線程提交任務(wù)的速度遠(yuǎn)遠(yuǎn)大于CachedThreadPool的處理速度，則CachedThreadPool會不斷地創(chuàng)建新線程來執(zhí)行任務(wù)，這樣有可能會導(dǎo)致系統(tǒng)耗盡CPU和內(nèi)存資源，所以在使用該線程池是，一定要注意控制并發(fā)的任務(wù)數(shù)，否則創(chuàng)建大量的線程可能導(dǎo)致嚴(yán)重的性能問題。

五、執(zhí)行過程

execute()方法執(zhí)行任務(wù)：
（1）execute(Runnable command)：
　　workerCount < corePoolSize：addWorker(command, true)
　　workerCount > corePoolSize && 任務(wù)可放入阻塞隊列：addWorker(null, false);
　　workerCount > BlockingQueue.size：addWorker(command, false)，如果失敗拒絕策略
（2）addWorker(Runnable firstTask, boolean core)：new一個worker線程來執(zhí)行任務(wù)
　　ctl + 1
　　workers.add(worker);
　　worker.start();
（3）worker.run()-runWorker(Worker w)：
　　Runnable task = w.firstTask;
　　while(task != null || (task = getTask()) != null) {
　　　　task.run();
　　}
　　執(zhí)行任務(wù)的run()方法
　　執(zhí)行完一個任務(wù)后，在阻塞隊列中取任務(wù)繼續(xù)執(zhí)行
（4）getTask()：workQueue.take();阻塞隊列中取任務(wù)

提交任務(wù)：

線程池根據(jù)業(yè)務(wù)不同的需求提供了兩種方式提交任務(wù)：Executor.execute()、ExecutorService.submit()。其中ExecutorService.submit()可以獲取該任務(wù)執(zhí)行的Future。

execute()

執(zhí)行流程如下：

（1）如果線程池當(dāng)前線程數(shù)小于corePoolSize，則調(diào)用addWorker創(chuàng)建新線程執(zhí)行任務(wù)，成功返回true，失敗執(zhí)行步驟2。

（2）如果線程池處于RUNNING狀態(tài)，則嘗試加入阻塞隊列，如果加入阻塞隊列成功，則嘗試進(jìn)行Double Check，如果加入失敗，則執(zhí)行步驟3。

　　如果加入阻塞隊列成功了，則會進(jìn)行一個Double Check的過程。Double Check過程的主要目的是判斷加入到阻塞隊里中的線程是否可以被執(zhí)行。如果線程池不是RUNNING狀態(tài)，則調(diào)用remove()方法從阻塞隊列中刪除該任務(wù)，然后調(diào)用reject()方法處理任務(wù)。否則需要確保還有線程執(zhí)行。

（3）如果線程池不是RUNNING狀態(tài)或者加入阻塞隊列失敗，則嘗試創(chuàng)建新線程直到maxPoolSize，如果失敗，則調(diào)用reject()方法運(yùn)行相應(yīng)的拒絕策略。

public void execute(Runnable command) {if (command == null)throw new NullPointerException();// 前面說的那個表示 “線程池狀態(tài)” 和 “線程數(shù)” 的整數(shù)int c = ctl.get();// 如果當(dāng)前線程數(shù)少于核心線程數(shù)，那么直接添加一個 worker 來執(zhí)行任務(wù)，// 創(chuàng)建一個新的線程，并把當(dāng)前任務(wù) command 作為這個線程的第一個任務(wù)(firstTask)if (workerCountOf(c) < corePoolSize) {// 添加任務(wù)成功，那么就結(jié)束了。提交任務(wù)嘛，線程池已經(jīng)接受了這個任務(wù)，這個方法也就可以返回了// 至于執(zhí)行的結(jié)果，到時候會包裝到 FutureTask 中。// 返回 false 代表線程池不允許提交任務(wù)if (addWorker(command, true))return;c = ctl.get();}// 到這里說明，要么當(dāng)前線程數(shù)大于等于核心線程數(shù)，要么剛剛 addWorker 失敗了// 如果線程池處于 RUNNING 狀態(tài)，把這個任務(wù)添加到任務(wù)隊列 workQueue 中if (isRunning(c) && workQueue.offer(command)) {/* 這里面說的是，如果任務(wù)進(jìn)入了 workQueue，我們是否需要開啟新的線程* 因為線程數(shù)在 [0, corePoolSize) 是無條件開啟新的線程* 如果線程數(shù)已經(jīng)大于等于 corePoolSize，那么將任務(wù)添加到隊列中，然后進(jìn)到這里*/int recheck = ctl.get();// 如果線程池已不處于 RUNNING 狀態(tài)，那么移除已經(jīng)入隊的這個任務(wù)，并且執(zhí)行拒絕策略if (! isRunning(recheck) && remove(command))reject(command);// 如果線程池還是 RUNNING 的，并且線程數(shù)為 0，那么開啟新的線程// 到這里，我們知道了，這塊代碼的真正意圖是：擔(dān)心任務(wù)提交到隊列中了，但是線程都關(guān)閉了else if (workerCountOf(recheck) == 0)addWorker(null, false);}// 如果 workQueue 隊列滿了，那么進(jìn)入到這個分支// 以 maximumPoolSize 為界創(chuàng)建新的 worker，// 如果失敗，說明當(dāng)前線程數(shù)已經(jīng)達(dá)到 maximumPoolSize，執(zhí)行拒絕策略else if (!addWorker(command, false))reject(command); }

addWorker

在這里需要好好理論addWorker中的參數(shù)，在execute()方法中，有三處調(diào)用了該方法：
第一次：workerCountOf(c) < corePoolSize ==> addWorker(command, true)，這個很好理解，當(dāng)然線程池的線程數(shù)量小于 corePoolSize ，則新建線程執(zhí)行任務(wù)即可，在執(zhí)行過程core == true，內(nèi)部與corePoolSize比較即可。
第二次：加入阻塞隊列進(jìn)行Double Check時，else if (workerCountOf(recheck) == 0) ==>addWorker(null, false)。如果線程池中的線程==0，按照道理應(yīng)該該任務(wù)應(yīng)該新建線程執(zhí)行任務(wù)，但是由于已經(jīng)該任務(wù)已經(jīng)添加到了阻塞隊列，那么就在線程池中新建一個空線程，然后從阻塞隊列中取線程即可。
第三次：線程池不是RUNNING狀態(tài)或者加入阻塞隊列失敗：else if (!addWorker(command, false))，這里core == fase，則意味著是與maximumPoolSize比較。

執(zhí)行流程：

（1）判斷當(dāng)前線程是否可以添加任務(wù)，如果可以則進(jìn)行下一步，否則return false；

rs >= SHUTDOWN ，表示當(dāng)前線程處于SHUTDOWN ，STOP、TIDYING、TERMINATED狀態(tài)

rs == SHUTDOWN , firstTask != null時不允許添加線程，因為線程處于SHUTDOWN 狀態(tài)，不允許添加任務(wù)

rs == SHUTDOWN , firstTask == null，但workQueue.isEmpty() == true，不允許添加線程，因為firstTask == null是為了添加一個沒有任務(wù)的線程然后再從workQueue中獲取任務(wù)的，如果workQueue == null，則說明添加的任務(wù)沒有任何意義。

（2）內(nèi)嵌循環(huán)，通過CAS worker + 1

（3）獲取主鎖mailLock，如果線程池處于RUNNING狀態(tài)獲取處于SHUTDOWN狀態(tài)且 firstTask == null，則將任務(wù)添加到workers Queue中，然后釋放主鎖mainLock，然后啟動線程，然后return true，如果中途失敗導(dǎo)致workerStarted= false，則調(diào)用addWorkerFailed()方法進(jìn)行處理。

// 第一個參數(shù)是準(zhǔn)備提交給這個線程執(zhí)行的任務(wù)，之前說了，可以為 null // 第二個參數(shù)為 true 代表使用核心線程數(shù) corePoolSize 作為創(chuàng)建線程的界線，也就說創(chuàng)建這個線程的時候， // 如果線程池中的線程總數(shù)已經(jīng)達(dá)到 corePoolSize，那么不能響應(yīng)這次創(chuàng)建線程的請求 // 如果是 false，代表使用最大線程數(shù) maximumPoolSize 作為界線 private boolean addWorker(Runnable firstTask, boolean core) {retry:for (;;) {int c = ctl.get();int rs = runStateOf(c);// 這個非常不好理解// 如果線程池已關(guān)閉，并滿足以下條件之一，那么不創(chuàng)建新的 worker：// 1. 線程池狀態(tài)大于 SHUTDOWN，其實(shí)也就是 STOP, TIDYING, 或 TERMINATED// 2. firstTask != null// 3. workQueue.isEmpty()// 簡單分析下：// 還是狀態(tài)控制的問題，當(dāng)線程池處于 SHUTDOWN 的時候，不允許提交任務(wù)，但是已有的任務(wù)繼續(xù)執(zhí)行// 當(dāng)狀態(tài)大于 SHUTDOWN 時，不允許提交任務(wù)，且中斷正在執(zhí)行的任務(wù)// 多說一句：如果線程池處于 SHUTDOWN，但是 firstTask 為 null，且 workQueue 非空，那么是允許創(chuàng)建 worker 的if (rs >= SHUTDOWN &&! (rs == SHUTDOWN &&firstTask == null &&! workQueue.isEmpty()))return false;for (;;) {int wc = workerCountOf(c);if (wc >= CAPACITY ||wc >= (core ? corePoolSize : maximumPoolSize))return false;// 如果成功，那么就是所有創(chuàng)建線程前的條件校驗都滿足了，準(zhǔn)備創(chuàng)建線程執(zhí)行任務(wù)了// 這里失敗的話，說明有其他線程也在嘗試往線程池中創(chuàng)建線程if (compareAndIncrementWorkerCount(c))break retry;// 由于有并發(fā)，重新再讀取一下 ctlc = ctl.get();// 正常如果是 CAS 失敗的話，進(jìn)到下一個里層的for循環(huán)就可以了// 可是如果是因為其他線程的操作，導(dǎo)致線程池的狀態(tài)發(fā)生了變更，如有其他線程關(guān)閉了這個線程池// 那么需要回到外層的for循環(huán)if (runStateOf(c) != rs)continue retry;// else CAS failed due to workerCount change; retry inner loop }}/* * 到這里，我們認(rèn)為在當(dāng)前這個時刻，可以開始創(chuàng)建線程來執(zhí)行任務(wù)了，* 因為該校驗的都校驗了，至于以后會發(fā)生什么，那是以后的事，至少當(dāng)前是滿足條件的*/// worker 是否已經(jīng)啟動boolean workerStarted = false;// 是否已將這個 worker 添加到 workers 這個 HashSet 中boolean workerAdded = false;Worker w = null;try {final ReentrantLock mainLock = this.mainLock;// 把 firstTask 傳給 worker 的構(gòu)造方法w = new Worker(firstTask);// 取 worker 中的線程對象，之前說了，Worker的構(gòu)造方法會調(diào)用 ThreadFactory 來創(chuàng)建一個新的線程final Thread t = w.thread;if (t != null) {// 這個是整個類的全局鎖，持有這個鎖才能讓下面的操作“順理成章”，// 因為關(guān)閉一個線程池需要這個鎖，至少我持有鎖的期間，線程池不會被關(guān)閉 mainLock.lock();try {int c = ctl.get();int rs = runStateOf(c);// 小于 SHUTTDOWN 那就是 RUNNING，這個自不必說，是最正常的情況// 如果等于 SHUTDOWN，前面說了，不接受新的任務(wù)，但是會繼續(xù)執(zhí)行等待隊列中的任務(wù)if (rs < SHUTDOWN ||(rs == SHUTDOWN && firstTask == null)) {// worker 里面的 thread 可不能是已經(jīng)啟動的if (t.isAlive())throw new IllegalThreadStateException();// 加到 workers 這個 HashSet 中 workers.add(w);int s = workers.size();// largestPoolSize 用于記錄 workers 中的個數(shù)的最大值// 因為 workers 是不斷增加減少的，通過這個值可以知道線程池的大小曾經(jīng)達(dá)到的最大值if (s > largestPoolSize)largestPoolSize = s;workerAdded = true;}} finally {mainLock.unlock();}// 添加成功的話，啟動這個線程if (workerAdded) {// 啟動線程 t.start();workerStarted = true;}}} finally {// 如果線程沒有啟動，需要做一些清理工作，如前面 workCount 加了 1，將其減掉if (! workerStarted)addWorkerFailed(w);}// 返回線程是否啟動成功return workerStarted;

Woker內(nèi)部類

從Worker的源碼中我們可以看到Woker繼承AQS，實(shí)現(xiàn)Runnable接口，所以可以認(rèn)為Worker既是一個可以執(zhí)行的任務(wù)，也可以達(dá)到獲取鎖釋放鎖的效果。這里繼承AQS主要是為了方便線程的中斷處理。這里注意兩個地方：構(gòu)造函數(shù)、run()。構(gòu)造函數(shù)主要是做三件事：1.設(shè)置同步狀態(tài)state為-1，同步狀態(tài)大于0表示就已經(jīng)獲取了鎖，2.設(shè)置將當(dāng)前任務(wù)task設(shè)置為firstTask，3.利用Worker本身對象this和ThreadFactory創(chuàng)建線程對象。

private final class Worker extends AbstractQueuedSynchronizerimplements Runnable {private static final long serialVersionUID = 6138294804551838833L;// task 的threadfinal Thread thread;// 運(yùn)行的任務(wù)task Runnable firstTask;volatile long completedTasks;Worker(Runnable firstTask) {//設(shè)置AQS的同步狀態(tài)private volatile int state，是一個計數(shù)器，大于0代表鎖已經(jīng)被獲取setState(-1);this.firstTask = firstTask;// 利用ThreadFactory和 Worker這個Runnable創(chuàng)建的線程對象this.thread = getThreadFactory().newThread(this);}// 任務(wù)執(zhí)行public void run() {runWorker(this);}}

runWorker

運(yùn)行流程：

（1）根據(jù)worker獲取要執(zhí)行的任務(wù)task，然后調(diào)用unlock()方法釋放鎖，這里釋放鎖的主要目的在于中斷，因為在new Worker時，設(shè)置的state為-1，調(diào)用unlock()方法可以將state設(shè)置為0，這里主要原因就在于interruptWorkers()方法只有在state >= 0時才會執(zhí)行；

（2）通過getTask()獲取執(zhí)行的任務(wù)，調(diào)用task.run()執(zhí)行，當(dāng)然在執(zhí)行之前會調(diào)用worker.lock()上鎖，執(zhí)行之后調(diào)用worker.unlock()放鎖；

（3）在任務(wù)執(zhí)行前后，可以根據(jù)業(yè)務(wù)場景自定義beforeExecute() 和 afterExecute()方法，則兩個方法在ThreadPoolExecutor中是空實(shí)現(xiàn)；

（4）如果線程執(zhí)行完成，則會調(diào)用getTask()方法從阻塞隊列中獲取新任務(wù)，如果阻塞隊列為空，則根據(jù)是否超時來判斷是否需要阻塞；

（5）task == null或者拋出異常（beforeExecute()、task.run()、afterExecute()均有可能）導(dǎo)致worker線程終止，則調(diào)用processWorkerExit()方法處理worker退出流程。

// 此方法由 worker 線程啟動后調(diào)用，這里用一個 while 循環(huán)來不斷地從等待隊列中獲取任務(wù)并執(zhí)行 // 前面說了，worker 在初始化的時候，可以指定 firstTask，那么第一個任務(wù)也就可以不需要從隊列中獲取 final void runWorker(Worker w) {// Thread wt = Thread.currentThread();// 該線程的第一個任務(wù)(如果有的話)Runnable task = w.firstTask;w.firstTask = null;w.unlock(); // allow interruptsboolean completedAbruptly = true;try {// 循環(huán)調(diào)用 getTask 獲取任務(wù)while (task != null || (task = getTask()) != null) {w.lock(); // 如果線程池狀態(tài)大于等于 STOP，那么意味著該線程也要中斷if ((runStateAtLeast(ctl.get(), STOP) ||(Thread.interrupted() &&runStateAtLeast(ctl.get(), STOP))) &&!wt.isInterrupted())wt.interrupt();try {// 這是一個鉤子方法，留給需要的子類實(shí)現(xiàn) beforeExecute(wt, task);Throwable thrown = null;try {// 到這里終于可以執(zhí)行任務(wù)了 task.run();} catch (RuntimeException x) {thrown = x; throw x;} catch (Error x) {thrown = x; throw x;} catch (Throwable x) {// 這里不允許拋出 Throwable，所以轉(zhuǎn)換為 Errorthrown = x; throw new Error(x);} finally {// 也是一個鉤子方法，將 task 和異常作為參數(shù)，留給需要的子類實(shí)現(xiàn) afterExecute(task, thrown);}} finally {// 置空 task，準(zhǔn)備 getTask 獲取下一個任務(wù)task = null;// 累加完成的任務(wù)數(shù)w.completedTasks++;// 釋放掉 worker 的獨(dú)占鎖 w.unlock();}}completedAbruptly = false;} finally {// 如果到這里，需要執(zhí)行線程關(guān)閉：// 1. 說明 getTask 返回 null，也就是說，這個 worker 的使命結(jié)束了，執(zhí)行關(guān)閉// 2. 任務(wù)執(zhí)行過程中發(fā)生了異常// 第一種情況，已經(jīng)在代碼處理了將 workCount 減 1，這個在 getTask 方法分析中會說// 第二種情況，workCount 沒有進(jìn)行處理，所以需要在 processWorkerExit 中處理// 限于篇幅，我不準(zhǔn)備分析這個方法了，感興趣的讀者請自行分析源碼 processWorkerExit(w, completedAbruptly);} }

getTask()

// 此方法有三種可能： // 1. 阻塞直到獲取到任務(wù)返回。我們知道，默認(rèn) corePoolSize 之內(nèi)的線程是不會被回收的， // 它們會一直等待任務(wù) // 2. 超時退出。keepAliveTime 起作用的時候，也就是如果這么多時間內(nèi)都沒有任務(wù)，那么應(yīng)該執(zhí)行關(guān)閉 // 3. 如果發(fā)生了以下條件，此方法必須返回 null: // - 池中有大于 maximumPoolSize 個 workers 存在(通過調(diào)用 setMaximumPoolSize 進(jìn)行設(shè)置) // - 線程池處于 SHUTDOWN，而且 workQueue 是空的，前面說了，這種不再接受新的任務(wù) // - 線程池處于 STOP，不僅不接受新的線程，連 workQueue 中的線程也不再執(zhí)行 private Runnable getTask() {boolean timedOut = false; // Did the last poll() time out? retry:for (;;) {int c = ctl.get();int rs = runStateOf(c);// 兩種可能// 1. rs == SHUTDOWN && workQueue.isEmpty()// 2. rs >= STOPif (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {// CAS 操作，減少工作線程數(shù) decrementWorkerCount();return null;}boolean timed; // Are workers subject to culling?for (;;) {int wc = workerCountOf(c);// 允許核心線程數(shù)內(nèi)的線程回收，或當(dāng)前線程數(shù)超過了核心線程數(shù)，那么有可能發(fā)生超時關(guān)閉timed = allowCoreThreadTimeOut || wc > corePoolSize;// 這里 break，是為了不往下執(zhí)行后一個 if (compareAndDecrementWorkerCount(c))// 兩個 if 一起看：如果當(dāng)前線程數(shù) wc > maximumPoolSize，或者超時，都返回 null// 那這里的問題來了，wc > maximumPoolSize 的情況，為什么要返回 null？// 換句話說，返回 null 意味著關(guān)閉線程。// 那是因為有可能開發(fā)者調(diào)用了 setMaximumPoolSize 將線程池的 maximumPoolSize 調(diào)小了if (wc <= maximumPoolSize && ! (timedOut && timed))break;if (compareAndDecrementWorkerCount(c))return null;c = ctl.get(); // Re-read ctl// compareAndDecrementWorkerCount(c) 失敗，線程池中的線程數(shù)發(fā)生了改變if (runStateOf(c) != rs)continue retry;// else CAS failed due to workerCount change; retry inner loop }// wc <= maximumPoolSize 同時沒有超時try {// 到 workQueue 中獲取任務(wù)Runnable r = timed ?workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :workQueue.take();if (r != null)return r;timedOut = true;} catch (InterruptedException retry) {// 如果此 worker 發(fā)生了中斷，采取的方案是重試// 解釋下為什么會發(fā)生中斷，這個讀者要去看 setMaximumPoolSize 方法，// 如果開發(fā)者將 maximumPoolSize 調(diào)小了，導(dǎo)致其小于當(dāng)前的 workers 數(shù)量，// 那么意味著超出的部分線程要被關(guān)閉。重新進(jìn)入 for 循環(huán)，自然會有部分線程會返回 nulltimedOut = false;}} }

processWorkerExit()

在runWorker()方法中，無論最終結(jié)果如何，都會執(zhí)行processWorkerExit()方法對worker進(jìn)行退出處理。

首先completedAbruptly的值來判斷是否需要對線程數(shù)-1處理，如果completedAbruptly == true，說明在任務(wù)運(yùn)行過程中出現(xiàn)了異常，那么需要進(jìn)行減1處理，否則不需要，因為減1處理在getTask()方法中處理了。然后從HashSet中移出該worker，過程需要獲取mainlock。然后調(diào)用tryTerminate()方法處理，該方法是對最后一個線程退出做終止線程池動作。如果線程池沒有終止，那么線程池需要保持一定數(shù)量的線程，則通過addWorker(null,false)新增一個空的線程。

private void processWorkerExit(Worker w, boolean completedAbruptly) {// true：用戶線程運(yùn)行異常,需要扣減// false：getTask方法中扣減線程數(shù)量if (completedAbruptly)decrementWorkerCount();// 獲取主鎖final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {completedTaskCount += w.completedTasks;// 從HashSet中移出worker workers.remove(w);} finally {mainLock.unlock();}// 有worker線程移除，可能是最后一個線程退出需要嘗試終止線程池 tryTerminate();int c = ctl.get();// 如果線程為running或shutdown狀態(tài)，即tryTerminate()沒有成功終止線程池，則判斷是否有必要一個workerif (runStateLessThan(c, STOP)) {// 正常退出，計算min：需要維護(hù)的最小線程數(shù)量if (!completedAbruptly) {// allowCoreThreadTimeOut 默認(rèn)false：是否需要維持核心線程的數(shù)量int min = allowCoreThreadTimeOut ? 0 : corePoolSize;// 如果min ==0 或者workerQueue為空，min = 1if (min == 0 && ! workQueue.isEmpty())min = 1;// 如果線程數(shù)量大于最少數(shù)量min，直接返回，不需要新增線程if (workerCountOf(c) >= min)return; // replacement not needed }// 添加一個沒有firstTask的workeraddWorker(null, false);}}

六、關(guān)閉線程池

tryTerminate()

當(dāng)線程池涉及到要移除worker時候都會調(diào)用tryTerminate()，該方法主要用于判斷線程池中的線程是否已經(jīng)全部移除了，如果是的話則關(guān)閉線程池。

final void tryTerminate() {for (;;) {int c = ctl.get();// 線程池處于Running狀態(tài)// 線程池已經(jīng)終止了// 線程池處于ShutDown狀態(tài)，但是阻塞隊列不為空if (isRunning(c) ||runStateAtLeast(c, TIDYING) ||(runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))return;// 執(zhí)行到這里，就意味著線程池要么處于STOP狀態(tài)，要么處于SHUTDOWN且阻塞隊列為空// 這時如果線程池中還存在線程，則會嘗試中斷線程if (workerCountOf(c) != 0) {// /線程池還有線程，但是隊列沒有任務(wù)了，需要中斷喚醒等待任務(wù)的線程// （runwoker的時候首先就通過w.unlock設(shè)置線程可中斷，getTask最后面的catch處理中斷） interruptIdleWorkers(ONLY_ONE);return;}final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {// 嘗試終止線程池if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {try {terminated();} finally {// 線程池狀態(tài)轉(zhuǎn)為TERMINATEDctl.set(ctlOf(TERMINATED, 0));termination.signalAll();}return;}} finally {mainLock.unlock();}}}

線程池ThreadPoolExecutor提供了shutdown()和shutDownNow()用于關(guān)閉線程池。

shutdown()：按過去執(zhí)行已提交任務(wù)的順序發(fā)起一個有序的關(guān)閉，但是不接受新任務(wù)。

public void shutdown() {final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {checkShutdownAccess();// 推進(jìn)線程狀態(tài) advanceRunState(SHUTDOWN);// 中斷空閑的線程 interruptIdleWorkers();// 交給子類實(shí)現(xiàn) onShutdown();} finally {mainLock.unlock();}tryTerminate();}

shutdownNow() :嘗試停止所有的活動執(zhí)行任務(wù)、暫停等待任務(wù)的處理，并返回等待執(zhí)行的任務(wù)列表。

shutdownNow會調(diào)用interruptWorkers()方法中斷所有線程，同時會調(diào)用drainQueue()方法返回等待執(zhí)行到任務(wù)列表。

public List<Runnable> shutdownNow() {List<Runnable> tasks;final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {checkShutdownAccess();advanceRunState(STOP);// 中斷所有線程 interruptWorkers();// 返回等待執(zhí)行的任務(wù)列表tasks = drainQueue();} finally {mainLock.unlock();}tryTerminate();return tasks;} private void interruptWorkers() {final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {for (Worker w : workers)w.interruptIfStarted();} finally {mainLock.unlock();}} private List<Runnable> drainQueue() {BlockingQueue<Runnable> q = workQueue;ArrayList<Runnable> taskList = new ArrayList<Runnable>();q.drainTo(taskList);if (!q.isEmpty()) {for (Runnable r : q.toArray(new Runnable[0])) {if (q.remove(r))taskList.add(r);}}return taskList;}

七、其他問題

1、任務(wù)拒絕策略

execute()方法中addWorker()失敗會調(diào)用reject(command) 來處理任務(wù)

線程池此時不能接受這個任務(wù)，所以需要執(zhí)行拒絕策略

此處的 handler 我們需要在構(gòu)造線程池的時候就傳入這個參數(shù)，它是 RejectedExecutionHandler 的實(shí)例。

RejectedExecutionHandler 在 ThreadPoolExecutor 中有四個已經(jīng)定義好的實(shí)現(xiàn)類可供我們直接使用，當(dāng)然，我們也可以實(shí)現(xiàn)自己的策略，不過一般也沒有必要。

final void reject(Runnable command) {// 執(zhí)行拒絕策略handler.rejectedExecution(command, this); }// 只要線程池沒有被關(guān)閉，那么由提交任務(wù)的線程自己來執(zhí)行這個任務(wù)。 public static class CallerRunsPolicy implements RejectedExecutionHandler {public CallerRunsPolicy() { }public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {if (!e.isShutdown()) {r.run();}} }// 不管怎樣，直接拋出 RejectedExecutionException 異常 // 這個是默認(rèn)的策略，如果我們構(gòu)造線程池的時候不傳相應(yīng)的 handler 的話，那就會指定使用這個 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());} }// 不做任何處理，直接忽略掉這個任務(wù) public static class DiscardPolicy implements RejectedExecutionHandler {public DiscardPolicy() { }public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {} }// 這個相對霸道一點(diǎn)，如果線程池沒有被關(guān)閉的話， // 把隊列隊頭的任務(wù)(也就是等待了最長時間的)直接扔掉，然后提交這個任務(wù)到等待隊列中 public static class DiscardOldestPolicy implements RejectedExecutionHandler {public DiscardOldestPolicy() { }public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {if (!e.isShutdown()) {e.getQueue().poll();e.execute(r);}} }

2、線程池中的線程初始化

默認(rèn)情況下，創(chuàng)建線程池之后，線程池中是沒有線程的，需要提交任務(wù)之后才會創(chuàng)建線程。

在實(shí)際中如果需要線程池創(chuàng)建之后立即創(chuàng)建線程，可以通過以下兩個方法辦到：

prestartCoreThread()：初始化一個核心線程；

prestartAllCoreThreads()：初始化所有核心線程

3、任務(wù)緩存隊列及排隊策略

workQueue的類型為BlockingQueue<Runnable>，通常可以取下面三種類型：

（1）ArrayBlockingQueue：基于數(shù)組的先進(jìn)先出隊列，此隊列創(chuàng)建時必須指定大小；

（2）LinkedBlockingQueue：基于鏈表的先進(jìn)先出隊列，如果創(chuàng)建時沒有指定此隊列大小，則默認(rèn)為Integer.MAX_VALUE；

（3）synchronousQueue：這個隊列比較特殊，它不會保存提交的任務(wù)，而是將直接新建一個線程來執(zhí)行新來的任務(wù)。

更多BlockingQueue參考?Java并發(fā)（十八）：阻塞隊列BlockingQueue

4、線程池容量的動態(tài)調(diào)整

ThreadPoolExecutor提供了動態(tài)調(diào)整線程池容量大小的方法：setCorePoolSize()和setMaximumPoolSize()，

setCorePoolSize：設(shè)置核心池大小

setMaximumPoolSize：設(shè)置線程池最大能創(chuàng)建的線程數(shù)目大小

當(dāng)上述參數(shù)從小變大時，ThreadPoolExecutor進(jìn)行線程賦值，還可能立即創(chuàng)建新的線程來執(zhí)行任務(wù)。

5、線程池的監(jiān)控

（1）通過線程池提供的參數(shù)進(jìn)行監(jiān)控。

　　taskCount：線程池需要執(zhí)行的任務(wù)數(shù)量。

　　completedTaskCount：線程池在運(yùn)行過程中已完成的任務(wù)數(shù)量。小于或等于taskCount。

　　largestPoolSize：線程池曾經(jīng)創(chuàng)建過的最大線程數(shù)量。通過這個數(shù)據(jù)可以知道線程池是否滿過。如等于線程池的最大大小，則表示線程池曾經(jīng)滿了。

　　getPoolSize:線程池的線程數(shù)量。如果線程池不銷毀的話，池里的線程不會自動銷毀，所以這個大小只增不減。

　　getActiveCount：獲取活動的線程數(shù)。

（2）通過擴(kuò)展線程池進(jìn)行監(jiān)控。

通過繼承線程池并重寫線程池的beforeExecute，afterExecute和terminated方法，我們可以在任務(wù)執(zhí)行前，執(zhí)行后和線程池關(guān)閉前干一些事情。如監(jiān)控任務(wù)的平均執(zhí)行時間，最大執(zhí)行時間和最小執(zhí)行時間等。這幾個方法在線程池里是空方法

八、線程池使用示例

為什么使用線程池？

1、降低資源消耗。通過重復(fù)利用已創(chuàng)建的線程降低線程創(chuàng)建和銷毀造成的消耗。

2、提高響應(yīng)速度。當(dāng)任務(wù)到達(dá)時，任務(wù)可以不需要的等到線程創(chuàng)建就能立即執(zhí)行。

3、提高線程的可管理性。線程是稀缺資源，如果無限制的創(chuàng)建，不僅會消耗系統(tǒng)資源，還會降低系統(tǒng)的穩(wěn)定性，使用線程池可以進(jìn)行統(tǒng)一的分配，調(diào)優(yōu)和監(jiān)控。但是要做到合理的利用線程池，必須對其原理了如指掌。

示例一：

class Test {public static void main(String[] args) {ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 10, 200,TimeUnit.MILLISECONDS, new ArrayBlockingQueue<Runnable>(5));for (int i = 0; i < 15; i++) {MyTask myTask = new MyTask(i);executor.execute(myTask);System.out.println("線程池中線程數(shù)目：" + executor.getPoolSize()+ "，隊列中等待執(zhí)行的任務(wù)數(shù)目：" + executor.getQueue().size()+ "，已執(zhí)行玩別的任務(wù)數(shù)目：" + executor.getCompletedTaskCount());}executor.shutdown();} }class MyTask implements Runnable {private int taskNum;public MyTask(int num) {this.taskNum = num;}@Overridepublic void run() {System.out.println("正在執(zhí)行task " + taskNum);try {Thread.currentThread().sleep(4000);} catch (InterruptedException e) {e.printStackTrace();}System.out.println("task " + taskNum + "執(zhí)行完畢");} }

執(zhí)行結(jié)果：

正在執(zhí)行task 0 線程池中線程數(shù)目：1，隊列中等待執(zhí)行的任務(wù)數(shù)目：0，已執(zhí)行玩別的任務(wù)數(shù)目：0 線程池中線程數(shù)目：2，隊列中等待執(zhí)行的任務(wù)數(shù)目：0，已執(zhí)行玩別的任務(wù)數(shù)目：0 正在執(zhí)行task 1 正在執(zhí)行task 2 線程池中線程數(shù)目：3，隊列中等待執(zhí)行的任務(wù)數(shù)目：0，已執(zhí)行玩別的任務(wù)數(shù)目：0 線程池中線程數(shù)目：4，隊列中等待執(zhí)行的任務(wù)數(shù)目：0，已執(zhí)行玩別的任務(wù)數(shù)目：0 正在執(zhí)行task 3 線程池中線程數(shù)目：5，隊列中等待執(zhí)行的任務(wù)數(shù)目：0，已執(zhí)行玩別的任務(wù)數(shù)目：0 正在執(zhí)行task 4 線程池中線程數(shù)目：5，隊列中等待執(zhí)行的任務(wù)數(shù)目：1，已執(zhí)行玩別的任務(wù)數(shù)目：0 線程池中線程數(shù)目：5，隊列中等待執(zhí)行的任務(wù)數(shù)目：2，已執(zhí)行玩別的任務(wù)數(shù)目：0 線程池中線程數(shù)目：5，隊列中等待執(zhí)行的任務(wù)數(shù)目：3，已執(zhí)行玩別的任務(wù)數(shù)目：0 線程池中線程數(shù)目：5，隊列中等待執(zhí)行的任務(wù)數(shù)目：4，已執(zhí)行玩別的任務(wù)數(shù)目：0 線程池中線程數(shù)目：5，隊列中等待執(zhí)行的任務(wù)數(shù)目：5，已執(zhí)行玩別的任務(wù)數(shù)目：0 線程池中線程數(shù)目：6，隊列中等待執(zhí)行的任務(wù)數(shù)目：5，已執(zhí)行玩別的任務(wù)數(shù)目：0 正在執(zhí)行task 10 線程池中線程數(shù)目：7，隊列中等待執(zhí)行的任務(wù)數(shù)目：5，已執(zhí)行玩別的任務(wù)數(shù)目：0 正在執(zhí)行task 11 線程池中線程數(shù)目：8，隊列中等待執(zhí)行的任務(wù)數(shù)目：5，已執(zhí)行玩別的任務(wù)數(shù)目：0 正在執(zhí)行task 12 線程池中線程數(shù)目：9，隊列中等待執(zhí)行的任務(wù)數(shù)目：5，已執(zhí)行玩別的任務(wù)數(shù)目：0 正在執(zhí)行task 13 線程池中線程數(shù)目：10，隊列中等待執(zhí)行的任務(wù)數(shù)目：5，已執(zhí)行玩別的任務(wù)數(shù)目：0 正在執(zhí)行task 14 task 0執(zhí)行完畢 正在執(zhí)行task 5 task 1執(zhí)行完畢 task 2執(zhí)行完畢 task 3執(zhí)行完畢 task 4執(zhí)行完畢 正在執(zhí)行task 9 正在執(zhí)行task 8 正在執(zhí)行task 7 正在執(zhí)行task 6 task 12執(zhí)行完畢 task 13執(zhí)行完畢 task 10執(zhí)行完畢 task 11執(zhí)行完畢 task 14執(zhí)行完畢 task 5執(zhí)行完畢 task 6執(zhí)行完畢 task 7執(zhí)行完畢 task 9執(zhí)行完畢 task 8執(zhí)行完畢 View Code

示例二：

需求：從數(shù)據(jù)庫中獲取url，并利用httpclient循環(huán)訪問url地址，并對返回結(jié)果進(jìn)行操作

分析：由于是循環(huán)的對多個url進(jìn)行訪問并獲取數(shù)據(jù)，為了執(zhí)行的效率，考慮使用多線程，url數(shù)量未知如果每個任務(wù)都創(chuàng)建一個線程將消耗大量的系統(tǒng)資源，最后決定使用線程池。

class GetMonitorDataService {private Logger logger = LoggerFactory.getLogger(GetMonitorDataService.class);@Resourceprivate MonitorProjectUrlMapper groupUrlMapper;@Resourceprivate MonitorDetailBatchInsertMapper monitorDetailBatchInsertMapper;public void sendData(){//調(diào)用dao查詢所有urlMonitorProjectUrlExample example=new MonitorProjectUrlExample();List<MonitorProjectUrl> list=groupUrlMapper.selectByExample(example);logger.info("此次查詢數(shù)據(jù)庫中監(jiān)控url個數(shù)為"+list.size());//獲取系統(tǒng)處理器個數(shù)，作為線程池數(shù)量int nThreads=Runtime.getRuntime().availableProcessors();//定義一個裝載多線程返回值的集合List<MonitorDetail> result= Collections.synchronizedList(new ArrayList<MonitorDetail>());//創(chuàng)建線程池，這里定義了一個創(chuàng)建線程池的工具類，避免了創(chuàng)建多個線程池，ThreadPoolFactoryUtil可以使用單例模式設(shè)計ExecutorService executorService = ThreadPoolFactoryUtil.getExecutorService(nThreads);//遍歷數(shù)據(jù)庫取出的urlif(list!=null&&list.size()>0) {for (MonitorProjectUrl monitorProjectUrl : list) {String url = monitorProjectUrl.getMonitorUrl();//創(chuàng)建任務(wù)ThreadTask threadTask = new ThreadTask(url, result);//執(zhí)行任務(wù) executorService.execute(threadTask);try {//等待直到所有任務(wù)完成 executorService.awaitTermination(Long.MAX_VALUE, TimeUnit.MINUTES);} catch (InterruptedException e) {e.printStackTrace();}}executorService.shutdown();//對數(shù)據(jù)進(jìn)行操作 saveData(result);}} }public class ThreadTask implements Runnable{//這里實(shí)現(xiàn)runnable接口private String url;private List<MonitorDetail> list;public ThreadTask(String url,List<MonitorDetail> list){this.url=url;this.list=list;}//把獲取的數(shù)據(jù)進(jìn)行處理 @Overridepublic void run() {MonitorDetail detail = HttpClientUtil.send(url, MonitorDetail.class);list.add(detail);}}

示例三：

public class FatureTest {//1、配置線程池private static ExecutorService es = Executors.newFixedThreadPool(20);//2、封裝響應(yīng)Featureclass BizResult{public String orderId;public String data;public String getOrderId() {return orderId;}public void setOrderId(String orderId) {this.orderId = orderId;}public String getData() {return data;}public void setData(String data) {this.data = data;}}//3、實(shí)現(xiàn)Callable接口class BizTask implements Callable {private String orderId;private Object data;//可以用其他方式private CountDownLatch countDownLatch;public BizTask(String orderId, Object data, CountDownLatch countDownLatch) {this.orderId = orderId;this.data = data;this.countDownLatch = countDownLatch;}@Overridepublic Object call() {try {//todo businessSystem.out.println("當(dāng)前線程Id = " + this.orderId);BizResult br = new BizResult();br.setOrderId(this.orderId);br.setData("some key about your business" + this.getClass());return br;}catch (Exception e){e.printStackTrace();}finally {//線程結(jié)束時，將計時器減一 countDownLatch.countDown();}return null;}}/*** 業(yè)務(wù)邏輯入口*/public List<Future> beginBusiness() throws InterruptedException {//模擬批量業(yè)務(wù)數(shù)據(jù)List<String> list = new ArrayList<>();for (int i = 0 ; i < 1000 ; i++) {list.add(String.valueOf(i));}//設(shè)置計數(shù)器CountDownLatch countDownLatch = new CountDownLatch(list.size());//接收多線程響應(yīng)結(jié)果List<Future> resultList = new ArrayList<>();//begin threadfor( int i = 0 ,size = list.size() ; i<size; i++){//todo something befor threadresultList.add(es.submit(new BizTask(list.get(i), null, countDownLatch)));}//wait finish countDownLatch.await();return resultList;}public static void main(String[] args) throws InterruptedException {FatureTest ft = new FatureTest();List<Future> futures = ft.beginBusiness();System.out.println("futures.size() = " + futures.size());//todo some operateSystem.out.println(" ==========================end========================= " );}}

?

?

參考資料 / 相關(guān)推薦：

深度解讀 java 線程池設(shè)計思想及源碼實(shí)現(xiàn)

【死磕Java并發(fā)】—–J.U.C之線程池：ThreadPoolExecutor

Java并發(fā)編程：線程池的使用

?聊聊并發(fā)（三）Java線程池的分析和使用

Java線程池架構(gòu)(一)原理和源碼解析

java線程池使用

?

轉(zhuǎn)載于:https://www.cnblogs.com/hexinwei1/p/10045955.html

總結(jié)

以上是生活随笔為你收集整理的Java并发（二十一）：线程池实现原理的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。