目錄介紹
- 01.實際開發問題
- 02.線程池的優勢
- 03.ThreadPoolExecutor參數
- 04.ThreadPoolExecutor使用
- 05.線程池執行流程
- 06.四種線程池類
- 07.execute和submit區別
- 08.線程池的使用技巧
01.實際開發問題
- 在我們的開發中經常會使用到多線程。例如在Android中，由於主線程的諸多限制，像網絡請求等一些耗時的操作我們必須在子線程中運行。
- 我們往往會通過new Thread來開啟一個子線程，待子線程操作完成以後通過Handler切換到主線程中運行。這麼以來我們無法管理我們所創建的子線程，並且無限制的創建子線程，它們相互之間競爭，很有可能由於佔用過多資源而導致死機或者OOM。所以在Java中為我們提供了線程池來管理我們所創建的線程。
02.線程池的優勢
- ①降低系統資源消耗，通過重用已存在的線程，降低線程創建和銷燬造成的消耗；
- ②提高系統響應速度，當有任務到達時，無需等待新線程的創建便能立即執行；
- ③方便線程併發數的管控，線程若是無限制的創建，不僅會額外消耗大量系統資源，更是佔用過多資源而阻塞系統或oom等狀況，從而降低系統的穩定性。線程池能有效管控線程，統一分配、調優，提供資源使用率；
- ④更強大的功能，線程池提供了定時、定期以及可控線程數等功能的線程池，使用方便簡單。
03.ThreadPoolExecutor
-
可以通過ThreadPoolExecutor來創建一個線程池。
ExecutorService service = new ThreadPoolExecutor(....); -
下面我們就來看一下ThreadPoolExecutor中的一個構造方法。
public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) - ThreadPoolExecutor參數含義
-
1.corePoolSize
- 線程池中的核心線程數，默認情況下，核心線程一直存活在線程池中，即便他們在線程池中處於閒置狀態。除非我們將ThreadPoolExecutor的allowCoreThreadTimeOut屬性設為true的時候，這時候處於閒置的核心線程在等待新任務到來時會有超時策略，這個超時時間由keepAliveTime來指定。一旦超過所設置的超時時間，閒置的核心線程就會被終止。
-
2.maximumPoolSize
- 線程池中所容納的最大線程數，如果活動的線程達到這個數值以後，後續的新任務將會被阻塞。包含核心線程數+非核心線程數。
-
3.keepAliveTime
- 非核心線程閒置時的超時時長，對於非核心線程，閒置時間超過這個時間，非核心線程就會被回收。只有對ThreadPoolExecutor的allowCoreThreadTimeOut屬性設為true的時候，這個超時時間才會對核心線程產生效果。
-
4.unit
- 用於指定keepAliveTime參數的時間單位。他是一個枚舉，可以使用的單位有天（TimeUnit.DAYS），小時（TimeUnit.HOURS），分鐘（TimeUnit.MINUTES），毫秒(TimeUnit.MILLISECONDS)，微秒(TimeUnit.MICROSECONDS, 千分之一毫秒)和毫微秒(TimeUnit.NANOSECONDS, 千分之一微秒);
-
5.workQueue
-
線程池中保存等待執行的任務的阻塞隊列。通過線程池中的execute方法提交的Runable對象都會存儲在該隊列中。我們可以選擇下面幾個阻塞隊列。我們還能夠通過實現BlockingQueue接口來自定義我們所需要的阻塞隊列。
| 阻塞隊列 | 說明 | | ------- | -------- | | ArrayBlockingQueue | 基於數組實現的有界的阻塞隊列,該隊列按照FIFO（先進先出）原則對隊列中的元素進行排序。| | LinkedBlockingQueue | 基於鏈表實現的阻塞隊列，該隊列按照FIFO（先進先出）原則對隊列中的元素進行排序。| | SynchronousQueue | 內部沒有任何容量的阻塞隊列。在它內部沒有任何的緩存空間。對於SynchronousQueue中的數據元素只有當我們試著取走的時候才可能存在。| | PriorityBlockingQueue | 具有優先級的無限阻塞隊列。|
-
-
6.threadFactory
- 線程工廠，為線程池提供新線程的創建。ThreadFactory是一個接口，裡面只有一個newThread方法。 默認為DefaultThreadFactory類。
-
7.handler
- 是RejectedExecutionHandler對象，而RejectedExecutionHandler是一個接口，裡面只有一個rejectedExecution方法。當任務隊列已滿並且線程池中的活動線程已經達到所限定的最大值或者是無法成功執行任務，這時候ThreadPoolExecutor會調用RejectedExecutionHandler中的rejectedExecution方法。在ThreadPoolExecutor中有四個內部類實現了RejectedExecutionHandler接口。在線程池中它默認是AbortPolicy，在無法處理新任務時拋出RejectedExecutionException異常。
- 下面是在ThreadPoolExecutor中提供的四個可選值。
-
我們也可以通過實現RejectedExecutionHandler接口來自定義我們自己的handler。如記錄日誌或持久化不能處理的任務。
| 可選值 | 說明 | | ----- | ------- | | CallerRunsPolicy | 只用調用者所在線程來運行任務。| | AbortPolicy | 直接拋出RejectedExecutionException異常。| | DiscardPolicy | 丟棄掉該任務，不進行處理。| | DiscardOldestPolicy | 丟棄隊列裡最近的一個任務，並執行當前任務。|
-
如下圖所示
04.ThreadPoolExecutor使用
-
如下所示
ExecutorService service = new ThreadPoolExecutor(5, 10, 10, TimeUnit.SECONDS, new LinkedBlockingQueue<>()); - 對於ThreadPoolExecutor有多個構造方法，對於上面的構造方法中的其他參數都採用默認值。可以通過execute和submit兩種方式來向線程池提交一個任務。
-
execute
- 當我們使用execute來提交任務時，由於execute方法沒有返回值，所以說我們也就無法判定任務是否被線程池執行成功。
service.execute(new Runnable() { public void run() { System.out.println("execute方式"); } }); -
submit
- 當我們使用submit來提交任務時,它會返回一個future,我們就可以通過這個future來判斷任務是否執行成功，還可以通過future的get方法來獲取返回值。如果子線程任務沒有完成，get方法會阻塞住直到任務完成，而使用get(long timeout, TimeUnit unit)方法則會阻塞一段時間後立即返回，這時候有可能任務並沒有執行完。
Future<Integer> future = service.submit(new Callable<Integer>() { @Override public Integer call() throws Exception { System.out.println("submit方式"); return 2; } }); try { Integer number = future.get(); } catch (ExecutionException e) { e.printStackTrace(); } -
線程池關閉
- 調用線程池的
shutdown()或shutdownNow()方法來關閉線程池 - shutdown原理：將線程池狀態設置成SHUTDOWN狀態，然後中斷所有沒有正在執行任務的線程。
- shutdownNow原理：將線程池的狀態設置成STOP狀態，然後中斷所有任務(包括正在執行的)的線程，並返回等待執行任務的列表。
- 中斷採用interrupt方法，所以無法響應中斷的任務可能永遠無法終止。 但調用上述的兩個關閉之一，isShutdown()方法返回值為true，當所有任務都已關閉，表示線程池關閉完成，則isTerminated()方法返回值為true。當需要立刻中斷所有的線程，不一定需要執行完任務，可直接調用shutdownNow()方法。
- 調用線程池的
05.線程池執行流程
-
大概的流程圖如下
-
文字描述如下
- ①如果在線程池中的線程數量沒有達到核心的線程數量，這時候就回啟動一個核心線程來執行任務。
- ②如果線程池中的線程數量已經超過核心線程數，這時候任務就會被插入到任務隊列中排隊等待執行。
- ③由於任務隊列已滿，無法將任務插入到任務隊列中。這個時候如果線程池中的線程數量沒有達到線程池所設定的最大值，那麼這時候就會立即啟動一個非核心線程來執行任務。
- ④如果線程池中的數量達到了所規定的最大值，那麼就會拒絕執行此任務，這時候就會調用RejectedExecutionHandler中的rejectedExecution方法來通知調用者。
06.四種線程池類
-
Java中四種具有不同功能常見的線程池。
- 他們都是直接或者間接配置ThreadPoolExecutor來實現他們各自的功能。這四種線程池分別是newFixedThreadPool,newCachedThreadPool,newScheduledThreadPool和newSingleThreadExecutor。這四個線程池可以通過Executors類獲取。
6.1 newFixedThreadPool
-
通過Executors中的newFixedThreadPool方法來創建，該線程池是一種線程數量固定的線程池。
ExecutorService service = Executors.newFixedThreadPool(4); -
在這個線程池中 所容納最大的線程數就是我們設置的核心線程數。
- 如果線程池的線程處於空閒狀態的話，它們並不會被回收，除非是這個線程池被關閉。如果所有的線程都處於活動狀態的話，新任務就會處於等待狀態，直到有線程空閒出來。
- 由於newFixedThreadPool只有核心線程，並且這些線程都不會被回收，也就是它能夠更快速的響應外界請求 。
-
從下面的newFixedThreadPool方法的實現可以看出，newFixedThreadPool只有核心線程，並且不存在超時機制，採用LinkedBlockingQueue，所以對於任務隊列的大小也是沒有限制的。
public static ExecutorService newFixedThreadPool(int nThreads) { return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>()); }
6.2 newCachedThreadPool
-
通過Executors中的newCachedThreadPool方法來創建。
public static ExecutorService newCachedThreadPool() { return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>()); } -
通過s上面的newCachedThreadPool方法在這裡我們可以看出它的 核心線程數為0， 線程池的最大線程數Integer.MAX_VALUE。而Integer.MAX_VALUE是一個很大的數，也差不多可以說 這個線程池中的最大線程數可以任意大。
- 當線程池中的線程都處於活動狀態的時候，線程池就會創建一個新的線程來處理任務。該線程池中的線程超時時長為60秒，所以當線程處於閒置狀態超過60秒的時候便會被回收。
- 這也就意味著若是整個線程池的線程都處於閒置狀態超過60秒以後，在newCachedThreadPool線程池中是不存在任何線程的，所以這時候它幾乎不佔用任何的系統資源。
- 對於newCachedThreadPool他的任務隊列採用的是SynchronousQueue，上面說到在SynchronousQueue內部沒有任何容量的阻塞隊列。SynchronousQueue內部相當於一個空集合，我們無法將一個任務插入到SynchronousQueue中。所以說在線程池中如果現有線程無法接收任務,將會創建新的線程來執行任務。
6.3 newScheduledThreadPool
-
通過Executors中的newScheduledThreadPool方法來創建。
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) { return new ScheduledThreadPoolExecutor(corePoolSize); } public ScheduledThreadPoolExecutor(int corePoolSize) { super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS, new DelayedWorkQueue()); } -
它的核心線程數是固定的，對於非核心線程幾乎可以說是沒有限制的，並且當非核心線程處於限制狀態的時候就會立即被回收。
- 創建一個可定時執行或週期執行任務的線程池：
ScheduledExecutorService service = Executors.newScheduledThreadPool(4); service.schedule(new Runnable() { public void run() { System.out.println(Thread.currentThread().getName()+"延遲三秒執行"); } }, 3, TimeUnit.SECONDS); service.scheduleAtFixedRate(new Runnable() { public void run() { System.out.println(Thread.currentThread().getName()+"延遲三秒後每隔2秒執行"); } }, 3, 2, TimeUnit.SECONDS);-
輸出結果：
>pool-1-thread-2延遲三秒後每隔2秒執行 ><br>pool-1-thread-1延遲三秒執行 ><br>pool-1-thread-1延遲三秒後每隔2秒執行 ><br>pool-1-thread-2延遲三秒後每隔2秒執行 ><br>pool-1-thread-2延遲三秒後每隔2秒執行
-
部分方法說明
-
schedule(Runnable command, long delay, TimeUnit unit)：延遲一定時間後執行Runnable任務； -
schedule(Callable callable, long delay, TimeUnit unit)：延遲一定時間後執行Callable任務； -
scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit)：延遲一定時間後，以間隔period時間的頻率週期性地執行任務； -
scheduleWithFixedDelay(Runnable command, long initialDelay, long delay,TimeUnit unit):與scheduleAtFixedRate()方法很類似，但是不同的是scheduleWithFixedDelay()方法的週期時間間隔是以上一個任務執行結束到下一個任務開始執行的間隔，而scheduleAtFixedRate()方法的週期時間間隔是以上一個任務開始執行到下一個任務開始執行的間隔，也就是這一些任務系列的觸發時間都是可預知的。
-
- ScheduledExecutorService功能強大，對於定時執行的任務，建議多采用該方法。
6.4 newSingleThreadExecutor
- 通過Executors中的newSingleThreadExecutor方法來創建，在這個線程池中只有一個核心線程，對於任務隊列沒有大小限制，也就意味著這一個任務處於活動狀態時，其他任務都會在任務隊列中排隊等候依次執行。
-
newSingleThreadExecutor將所有的外界任務統一到一個線程中支持，所以在這個任務執行之間我們不需要處理線程同步的問題。
public static ExecutorService newSingleThreadExecutor() { return new FinalizableDelegatedExecutorService (new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>())); }
07.execute和submit區別
-
先思考一個問題
- 為了保證項目中線程數量不會亂飆升，不好管理，我們會使用線程池，保證線程在我們的管理之下。
- 我們也經常說：使用線程池複用線程。那麼問題是：線程池中的線程是如何複用的？是執行完成後銷燬，再新建幾個放那；還是始終是那幾個線程（針對 coreSize 線程）。
-
execute和submit
- 調用線程池的execute方法(ExecutorService的submit方法最終也是調用execute)傳進去的Runnable，並不會直接以new Thread(runnable).start()的方式來執行，而是通過一個正在運行的線程來調用我們傳進去的Runnable的run方法的。
- 那麼，這個正在運行的線程，在執行完傳進去的Runnable的run方法後會銷燬嗎？看情況。
-
大部分場景下，我們都是通過Executors的newXXX方法來創建線程池的，就拿newCachedThreadPool來說：
public static ExecutorService newCachedThreadPool() { return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>()); }- 看第三個參數(keepAliveTime)：60L，後面的單位是秒，也就是說，newCachedThreadPool方法返回的線程池，它的工作線程(也就是用來調用Runnable的run方法的線程)的空閒等待時長為60秒，如果超過了60秒沒有獲取到新的任務，那麼這個工作線程就會結束。如果在60秒內接到了新的任務，那麼它會在新任務結束後重新等待。
-
還有另一種常用的線程池，通過newFixedThreadPool方法創建的：
public static ExecutorService newFixedThreadPool(int nThreads) { return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>()); }- 它跟上面的newCachedThreadPool方法一樣，創建的都是ThreadPoolExecutor的對象，只是參數不同而已。
可以看到第三個參數設置成了0，這就說明，如果當前工作線程數 > corePoolSize時，並且有工作線程在執行完上一個任務後沒拿到新的任務，那麼這個工作線程就會立即結束。
再看第二個參數(maximumPoolSize)，它設置成了跟corePoolSize一樣大，也就是說當前工作線程數 永遠不會大於 corePoolSize了，這樣的話，即使有工作線程是空閒的，也不會主動結束，會一直等待下一個任務的到來。
- 它跟上面的newCachedThreadPool方法一樣，創建的都是ThreadPoolExecutor的對象，只是參數不同而已。
-
ThreadPoolExecutor分析
- 來探究一下ThreadPoolExecutor是如何管理線程的，先來看精簡後的execute方法：
- 邏輯很清晰：當execute方法被調用時，如果當前工作線程 < corePoolSize(上面ThreadPoolExecutor構造方法的第一個參數)的話，就會創建新的線程，否則加入隊列。加入隊列後如果沒有工作線程在運行，也會創建一個。
private final BlockingQueue<Runnable> workQueue; public void execute(Runnable command) { int c = ctl.get(); //當前工作線程還沒滿 if (workerCountOf(c) < corePoolSize) { //可以創建新的工作線程來執行這個任務 if (addWorker(command, true)){ //添加成功直接返回 return; } } //如果工作線程滿了的話，會加入到阻塞隊列中 if (workQueue.offer(command)) { int recheck = ctl.get(); //加入到隊列之後，如果當前沒有工作線程，那麼就會創建一個工作線程 if (workerCountOf(recheck) == 0) addWorker(null, false); } }-
接著看它是怎麼創建新線程的：
- 主要操作是再次檢查，然後創建Worker對象，並且把worker對象店家到HashSet集合中，最後啟動工作線程。
private final HashSet<Worker> workers = new HashSet<>(); private boolean addWorker(Runnable firstTask, boolean core) { //再次檢查 int wc = workerCountOf(c); if (wc >= CAPACITY || wc >= corePoolSize) return false; boolean workerStarted = false; Worker w = null; //創建Worker對象 w = new Worker(firstTask); //添加到集合中 workers.add(w); final Thread t = w.thread; //啟動工作線程 t.start(); workerStarted = true; return workerStarted; }-
看看Worker裡面是怎麼樣的：
- 可以看到，這個Worker也是一個Runnable。構造方法裡面還創建了一個Thread，這個Thread對象，對應了上面addWorker方法啟動的那個thread。
private final class Worker extends AbstractQueuedSynchronizer implements Runnable { final Thread thread; Runnable firstTask; Worker(Runnable firstTask) { this.firstTask = firstTask; this.thread = getThreadFactory().newThread(this); } public void run() { runWorker(this); } }-
再看Worker類中的run方法，它調用了runWorker，並把自己傳了進去：
- Worker裡面的firstTask，就是我們通過execute方法傳進去的Runnable，可以看到它會在這個方法裡面被執行。
- 執行完成之後，接著就會通過getTask方法嘗試從等待隊列中(上面的workQueue)獲取下一個任務，如果getTask方法返回null的話，那麼這個工作線程就會結束。
final void runWorker(Worker w) { Runnable task = w.firstTask; w.firstTask = null; while (task != null || (task = getTask()) != null) { try { task.run(); } finally { task = null; w.completedTasks++; } } }- 最後看看runWorker方法中的getTask方法
private Runnable getTask() { boolean timedOut = false; // Did the last poll() time out? for (; ; ) { int c = ctl.get(); int wc = workerCountOf(c); //如果當前工作線程數大於指定的corePoolSize的話，就要視情況結束工作線程 boolean timed = wc > corePoolSize; //(當前工作線程數 > 指定的最大線程數 || (工作線程數 > 指定的核心線程數 && 上一次被標記超時了)) && (當前工作線程數有2個以上 || 等待隊列現在是空的) if ((wc > maximumPoolSize || (timed && timedOut)) && (wc > 1 || workQueue.isEmpty())) { return null; } //如果當前工作線程數大於指定的corePoolSize，就看能不能在keepAliveTime時間內獲取到新任務 //如果線程數沒有 > corePoolSize的話，就會一直等待 Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take(); if (r != null) return r; //沒能在keepAliveTime時間內獲取到新任務，標記已超時 timedOut = true; } }
08.線程池的使用技巧
-
需要針對具體情況而具體處理，不同的任務類別應採用不同規模的線程池，任務類別可劃分為CPU密集型任務、IO密集型任務和混合型任務。(N代表CPU個數)
| 任務類別 | 說明 | | ------ | ----------- | | CPU密集型任務 | 線程池中線程個數應儘量少，如配置N+1個線程的線程池。| | IO密集型任務 | 由於IO操作速度遠低於CPU速度，那麼在運行這類任務時，CPU絕大多數時間處於空閒狀態，那麼線程池可以配置儘量多些的線程，以提高CPU利用率，如2*N。| | 混合型任務 | 可以拆分為CPU密集型任務和IO密集型任務，當這兩類任務執行時間相差無幾時，通過拆分再執行的吞吐率高於串行執行的吞吐率，但若這兩類任務執行時間有數據級的差距，那麼沒有拆分的意義。 |