摘要:源碼分析創建可緩沖的線程池。源碼分析使用創建線程池源碼分析的構造函數構造函數參數核心線程數大小,當線程數,會創建線程執行最大線程數,當線程數的時候,會把放入中保持存活時間,當線程數大于的空閑線程能保持的最大時間。
之前創建線程的時候都是用的 newCachedThreadPoo,newFixedThreadPool,newScheduledThreadPool,newSingleThreadExecutor 這四個方法。
當然 Executors 也是用不同的參數去 new ThreadPoolExecutor 實現的,本文先分析前四種線程創建方式,后在分析 new ThreadPoolExecutor 創建方式
由于使用了LinkedBlockingQueue所以maximumPoolSize沒用,當corePoolSize滿了之后就加入到LinkedBlockingQueue隊列中。
每當某個線程執行完成之后就從LinkedBlockingQueue隊列中取一個。
所以這個是創建固定大小的線程池。
源碼分析
public static ExecutorService newFixedThreadPool(int nThreads) { return new ThreadPoolExecutor( nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue2.newSingleThreadPool()()); }
創建線程數為1的線程池,由于使用了LinkedBlockingQueue所以maximumPoolSize 沒用,corePoolSize為1表示線程數大小為1,滿了就放入隊列中,執行完了就從隊列取一個。
源碼分析
public static ExecutorService newSingleThreadExecutor() { return new Executors.FinalizableDelegatedExecutorService ( new ThreadPoolExecutor( 1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue3.newCachedThreadPool()()) ); } public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue workQueue) { this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, Executors.defaultThreadFactory(), defaultHandler); }
創建可緩沖的線程池。沒有大小限制。由于corePoolSize為0所以任務會放入SynchronousQueue隊列中,SynchronousQueue只能存放大小為1,所以會立刻新起線程,由于maxumumPoolSize為Integer.MAX_VALUE所以可以認為大小為2147483647。受內存大小限制。
源碼分析
public static ExecutorService newCachedThreadPool() { return new ThreadPoolExecutor( 0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue使用 ThreadPoolExecutor 創建線程池()); } public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue workQueue) { this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, Executors.defaultThreadFactory(), defaultHandler); }
源碼分析 ,ThreadPoolExecutor 的構造函數
public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue構造函數參數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; }
1、corePoolSize 核心線程數大小,當線程數 < corePoolSize ,會創建線程執行 runnable
2、maximumPoolSize 最大線程數, 當線程數 >= corePoolSize的時候,會把 runnable 放入 workQueue中
3、keepAliveTime 保持存活時間,當線程數大于corePoolSize的空閑線程能保持的最大時間。
4、unit 時間單位
5、workQueue 保存任務的阻塞隊列
6、threadFactory 創建線程的工廠
7、handler 拒絕策略
任務執行順序1、當線程數小于 corePoolSize時,創建線程執行任務。
2、當線程數大于等于 corePoolSize并且 workQueue 沒有滿時,放入workQueue中
3、線程數大于等于 corePoolSize并且當 workQueue 滿時,新任務新建線程運行,線程總數要小于 maximumPoolSize
4、當線程總數等于 maximumPoolSize 并且 workQueue 滿了的時候執行 handler 的 rejectedExecution。也就是拒絕策略。
JDK7提供了7個阻塞隊列。(也屬于并發容器)
ArrayBlockingQueue :一個由數組結構組成的有界阻塞隊列。
LinkedBlockingQueue :一個由鏈表結構組成的有界阻塞隊列。
PriorityBlockingQueue :一個支持優先級排序的無界阻塞隊列。
DelayQueue:一個使用優先級隊列實現的無界阻塞隊列。
SynchronousQueue:一個不存儲元素的阻塞隊列。
LinkedTransferQueue:一個由鏈表結構組成的無界阻塞隊列。
LinkedBlockingDeque:一個由鏈表結構組成的雙向阻塞隊列。
什么是阻塞隊列?阻塞隊列是一個在隊列基礎上又支持了兩個附加操作的隊列。
2個附加操作:
支持阻塞的插入方法:隊列滿時,隊列會阻塞插入元素的線程,直到隊列不滿。
支持阻塞的移除方法:隊列空時,獲取元素的線程會等待隊列變為非空。
阻塞隊列常用于生產者和消費者的場景,生產者是向隊列里添加元素的線程,消費者是從隊列里取元素的線程。簡而言之,阻塞隊列是生產者用來存放元素、消費者獲取元素的容器。
幾個方法在阻塞隊列不可用的時候,上述2個附加操作提供了四種處理方法
方法處理方式 | 拋出異常 | 返回特殊值 | 一直阻塞 | 超時退出 |
---|---|---|---|---|
插入方法 | add(e) | offer(e) | put(e) | offer(e,time,unit) |
移除方法 | remove() | poll() | take() | poll(time,unit) |
檢查方法 | element() | peek() | 不可用 | 不可用 |
JDK 7 提供了7個阻塞隊列,如下
1、ArrayBlockingQueue 數組結構組成的有界阻塞隊列。
此隊列按照先進先出(FIFO)的原則對元素進行排序,但是默認情況下不保證線程公平的訪問隊列,即如果隊列滿了,那么被阻塞在外面的線程對隊列訪問的順序是不能保證線程公平(即先阻塞,先插入)的。
2、LinkedBlockingQueue一個由鏈表結構組成的有界阻塞隊列
此隊列按照先出先進的原則對元素進行排序
3、PriorityBlockingQueue支持優先級的無界阻塞隊列
4、DelayQueue支持延時獲取元素的無界阻塞隊列,即可以指定多久才能從隊列中獲取當前元素
5、SynchronousQueue不存儲元素的阻塞隊列,每一個put必須等待一個take操作,否則不能繼續添加元素。并且他支持公平訪問隊列。
6、LinkedTransferQueue由鏈表結構組成的無界阻塞TransferQueue隊列。相對于其他阻塞隊列,多了tryTransfer和transfer方法
transfer方法
如果當前有消費者正在等待接收元素(take或者待時間限制的poll方法),transfer可以把生產者傳入的元素立刻傳給消費者。如果沒有消費者等待接收元素,則將元素放在隊列的tail節點,并等到該元素被消費者消費了才返回。
tryTransfer方法
用來試探生產者傳入的元素能否直接傳給消費者。,如果沒有消費者在等待,則返回false。和上述方法的區別是該方法無論消費者是否接收,方法立即返回。而transfer方法是必須等到消費者消費了才返回。
7、LinkedBlockingDeque鏈表結構的雙向阻塞隊列,優勢在于多線程入隊時,減少一半的競爭。
四個拒絕策略ThreadPoolExecutor默認有四個拒絕策略:
1、ThreadPoolExecutor.AbortPolicy() 直接拋出異常RejectedExecutionException
2、ThreadPoolExecutor.CallerRunsPolicy() 直接調用run方法并且阻塞執行
3、ThreadPoolExecutor.DiscardPolicy() 直接丟棄后來的任務
4、ThreadPoolExecutor.DiscardOldestPolicy() 丟棄在隊列中隊首的任務
當然可以自己繼承RejectedExecutionHandler來寫拒絕策略.
TestThreadPoolExecutor 示例 TestThreadPoolExecutor.javapackage io.ymq.thread.TestThreadPoolExecutor; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; /** * 描述: * * @author yanpenglei * @create 2017-10-12 15:39 **/ public class TestThreadPoolExecutor { public static void main(String[] args) { long currentTimeMillis = System.currentTimeMillis(); // 構造一個線程池 ThreadPoolExecutor threadPool = new ThreadPoolExecutor(5, 6, 3, TimeUnit.SECONDS, new ArrayBlockingQueueThreadPoolTask.java(3) ); for (int i = 1; i <= 10; i++) { try { String task = "task=" + i; System.out.println("創建任務并提交到線程池中:" + task); threadPool.execute(new ThreadPoolTask(task)); Thread.sleep(100); } catch (Exception e) { e.printStackTrace(); } } try { //等待所有線程執行完畢當前任務。 threadPool.shutdown(); boolean loop = true; do { //等待所有線程執行完畢當前任務結束 loop = !threadPool.awaitTermination(2, TimeUnit.SECONDS);//等待2秒 } while (loop); if (loop != true) { System.out.println("所有線程執行完畢"); } } catch (InterruptedException e) { e.printStackTrace(); } finally { System.out.println("耗時:" + (System.currentTimeMillis() - currentTimeMillis)); } } }
package io.ymq.thread.TestThreadPoolExecutor; import java.io.Serializable; /** * 描述: * * @author yanpenglei * @create 2017-10-12 15:40 **/ public class ThreadPoolTask implements Runnable, Serializable { private Object attachData; ThreadPoolTask(Object tasks) { this.attachData = tasks; } public void run() { try { System.out.println("開始執行任務:" + attachData + "任務,使用的線程池,線程名稱:" + Thread.currentThread().getName()); System.out.println(); } catch (Exception e) { e.printStackTrace(); } attachData = null; } }
遇到java.util.concurrent.RejectedExecutionException
第一
你的線程池 ThreadPoolExecutor 顯示的 shutdown() 之后,再向線程池提交任務的時候。 如果你配置的拒絕策略是 AbortPolicy 的話,這個異常就會拋出來。
第二
當你設置的任務緩存隊列過小的時候,或者說, 你的線程池里面所有的線程都在干活(線程數== maxPoolSize),并且你的任務緩存隊列也已經充滿了等待的隊列, 這個時候,你再向它提交任務,則會拋出這個異常。
響應
可以看到線程 pool-1-thread-1 到5 循環使用
創建任務并提交到線程池中:task=1 開始執行任務:task=1任務,使用的線程池,線程名稱:pool-1-thread-1 創建任務并提交到線程池中:task=2 開始執行任務:task=2任務,使用的線程池,線程名稱:pool-1-thread-2 創建任務并提交到線程池中:task=3 開始執行任務:task=3任務,使用的線程池,線程名稱:pool-1-thread-3 創建任務并提交到線程池中:task=4 開始執行任務:task=4任務,使用的線程池,線程名稱:pool-1-thread-4 創建任務并提交到線程池中:task=5 開始執行任務:task=5任務,使用的線程池,線程名稱:pool-1-thread-5 創建任務并提交到線程池中:task=6 開始執行任務:task=6任務,使用的線程池,線程名稱:pool-1-thread-1 創建任務并提交到線程池中:task=7 開始執行任務:task=7任務,使用的線程池,線程名稱:pool-1-thread-2 創建任務并提交到線程池中:task=8 開始執行任務:task=8任務,使用的線程池,線程名稱:pool-1-thread-3 創建任務并提交到線程池中:task=9 開始執行任務:task=9任務,使用的線程池,線程名稱:pool-1-thread-4 創建任務并提交到線程池中:task=10 開始執行任務:task=10任務,使用的線程池,線程名稱:pool-1-thread-5 所有線程執行完畢 耗時:1015測試代碼
github https://github.com/souyunku/ymq-example/tree/master/ymq-thread
Contact作者:鵬磊
出處:http://www.ymq.io
Email:admin@souyunku.com
版權歸作者所有,轉載請注明出處
Wechat:關注公眾號,搜云庫,專注于開發技術的研究與知識分享
文章版權歸作者所有,未經允許請勿轉載,若此文章存在違規行為,您可以聯系管理員刪除。
轉載請注明本文地址:http://specialneedsforspecialkids.com/yun/67766.html
摘要:創建方法最大線程數即源碼單線程化的線程池有且僅有一個工作線程執行任務所有任務按照指定順序執行,即遵循隊列的入隊出隊規則創建方法源碼還有一個結合了和,就不介紹了,基本不用。 *本篇文章已授權微信公眾號 guolin_blog (郭霖)獨家發布 為什么用線程池 創建/銷毀線程伴隨著系統開銷,過于頻繁的創建/銷毀線程,會很大程度上影響處理效率 >例如: > >記創建線程消耗時間T1,執行...
摘要:本文主要內容為簡單總結中線程池的相關信息。方法簇方法簇用于創建固定線程數的線程池。三種常見線程池的對比上文總結了工具類創建常見線程池的方法,現對三種線程池區別進行比較。 概述 線程可認為是操作系統可調度的最小的程序執行序列,一般作為進程的組成部分,同一進程中多個線程可共享該進程的資源(如內存等)。在單核處理器架構下,操作系統一般使用分時的方式實現多線程;在多核處理器架構下,多個線程能夠...
摘要:線程池常見實現線程池一般包含三個主要部分調度器決定由哪個線程來執行任務執行任務所能夠的最大耗時等線程隊列存放并管理著一系列線程這些線程都處于阻塞狀態或休眠狀態任務隊列存放著用戶提交的需要被執行的任務一般任務的執行的即先提交的任務先被執行調度 線程池常見實現 線程池一般包含三個主要部分: 調度器: 決定由哪個線程來執行任務, 執行任務所能夠的最大耗時等 線程隊列: 存放并管理著一系列線...
閱讀 2942·2023-04-26 01:32
閱讀 1541·2021-09-13 10:37
閱讀 2278·2019-08-30 15:56
閱讀 1670·2019-08-30 14:00
閱讀 3043·2019-08-30 12:44
閱讀 1961·2019-08-26 12:20
閱讀 1056·2019-08-23 16:29
閱讀 3228·2019-08-23 14:44