您好,登錄后才能下訂單哦!
這篇文章主要介紹“怎么快速了解Java線程池”,在日常操作中,相信很多人在怎么快速了解Java線程池問題上存在疑惑,小編查閱了各式資料,整理出簡單好用的操作方法,希望對大家解答”怎么快速了解Java線程池”的疑惑有所幫助!接下來,請跟著小編一起來學習吧!
老王 是個深耕在帝都的一線碼農,辛苦一年掙了點錢,想把錢存儲到銀行卡里,錢銀行卡辦理遇到了如下的遭遇
直接辦理
了。坐著等
辦理去了。解決策略
。
直接告知人太多不給你辦理了。 看到 老王 就來氣,也不給不辦理也不讓他走。 經理讓 老王 取嘗試跟座位席中最前面的人聊一聊看是否可以加塞,可以就辦理,不可以還是被踢走。 經理直接跟 老王 說誰讓你來的你找誰去我這辦理不了。
上面的這個流程幾乎就跟JDK線程池的大致流程類似,
營業中的3個窗口對應核心線程池數:corePoolSize 銀行總的營業窗口數對應:maximumPoolSize 打開的臨時窗口在多少時間內無人辦理則關閉對應:unit 銀行里的等待座椅就是等待隊列:workQueue 無法辦理的時候銀行給出的解決方法對應:RejectedExecutionHandler threadFactory 該參數在JDK中是 線程工廠,用來創建線程對象,一般不會動。
5分鐘線程池的核心工作流程講解完畢,更細節的知識看下面。
簡單理解就是 預先創建好一定數量的線程對象,存入緩沖池中,需要用的時候直接從緩沖池中取出,用完之后不要銷毀,還回到緩沖池中。
提高線程的利用率,降低資源的消耗。 提高響應速度,線程的創建時間為T1,執行時間T2,銷毀時間T3,用線程池可以免去T1和T3的時間。 便于統一管理線程對象 可控制最大并發數
如果先不看線程池源碼讓我們自己手動實現一個線程池你可以考慮到幾個重要點?
有若干個初始化好的線程數組來充當線程池。 線程池要去一個 等待的任務隊列 中去拿任務。
簡單來說就是初始化N個線程充當線程池然后一起去阻塞隊列中進行阻塞take,新添加的任務都通過put將任務追加到任務隊列,關于任務隊列的講解看這blog
public class MyThreadPool2 {
// 線程池中默認線程的個數為5
private static int WORK_NUM = 5;
// 隊列默認任務個數為100 來不及保存任務
private static int TASK_COUNT = 100;
// 工作線程組
private WorkThread[] workThreads;
// 任務隊列,作為一個緩沖
private final BlockingQueue<Runnable> taskQueue;
//用戶在構造這個池,希望的啟動的線程數
private final int worker_num;
// 創建具有默認線程個數的線程池
public MyThreadPool2() {
this(WORK_NUM, TASK_COUNT);
}
// 創建線程池,worker_num為線程池中工作線程的個數
public MyThreadPool2(int worker_num, int taskCount) {
if (worker_num <= 0) worker_num = WORK_NUM;
if (taskCount <= 0) taskCount = TASK_COUNT;
this.worker_num = worker_num;
taskQueue = new ArrayBlockingQueue<>(taskCount);
workThreads = new WorkThread[worker_num];
for (int i = 0; i < worker_num; i++) {
workThreads[i] = new WorkThread();
workThreads[i].start();
}
Runtime.getRuntime().availableProcessors();
}
// 執行任務,其實只是把任務加入任務隊列,什么時候執行有線程池管理器決定
public void execute(Runnable task) {
try {
taskQueue.put(task);// 阻塞 放置任務
} catch (InterruptedException e) {
e.printStackTrace();
}
}
// 銷毀線程池,該方法保證在所有任務都完成的情況下才銷毀所有線程,否則等待任務完成才銷毀
public void destroy() {
// 工作線程停止工作,且置為null
System.out.println("準備關閉線程池");
for (int i = 0; i < worker_num; i++) {
workThreads[i].stopWorker();
workThreads[i] = null;//help gc
}
taskQueue.clear();// 清空任務隊列
}
// 覆蓋toString方法,返回線程池信息:工作線程個數和已完成任務個數
@Override
public String toString() {
return "線程池大小 :" + worker_num + " 等待執行任務個數:" + taskQueue.size();
}
//內部類,工作線程
private class WorkThread extends Thread {
@Override
public void run() {
Runnable r = null;
try {
while (!isInterrupted()) {
r = taskQueue.take();//阻塞獲得任務
if (r != null) {
System.out.println(getId() + " 準備執行 :" + r);
r.run();
}
r = null; //help gc;
}
} catch (Exception e) {
// TODO: handle exception
}
}
public void stopWorker() {
interrupt();
}
}
}
public class TestMyThreadPool {
public static void main(String[] args) throws InterruptedException {
// 創建3個線程的線程池
MyThreadPool2 t = new MyThreadPool2(3, 5);
t.execute(new MyTask("testA"));
t.execute(new MyTask("testB"));
t.execute(new MyTask("testC"));
t.execute(new MyTask("testD"));
t.execute(new MyTask("testE"));
System.out.println(t);
Thread.sleep(10000);
t.destroy();// 所有線程都執行完成 才destory
System.out.println(t);
}
// 任務類
static class MyTask implements Runnable {
private String name;
private Random r = new Random();
public MyTask(String name) {
this.name = name;
}
public String getName() {
return name;
}
@Override
public void run() {// 執行任務
try {
Thread.sleep(r.nextInt(1000) + 2000); //隨機休眠
} catch (InterruptedException e) {
System.out.println(Thread.currentThread().getId() + " 被打斷:"
+ Thread.currentThread().isInterrupted());
}
System.out.println("任務 " + name + " 完成");
}
}
}
ThreadPoolExecutor
是JDK中所有線程池實現類的父類,構造函數有多個入參通過靈活的組合來實現線程池的初始化,核心構造函數如下。
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler)
此值是用來初始化線程池中核心線程數,當線程池中線程池數<
corePoolSize
時,系統默認是添加一個任務才創建一個線程池。可以通過調用prestartAllCoreThreads
方法一次性的啟動corePoolSize
個數的線程。當線程數 = corePoolSize時,新任務會追加到workQueue中。
maximumPoolSize
表示允許的最大線程數 = (非核心線程數+核心線程數),當BlockingQueue
也滿了,但線程池中總線程數 <maximumPoolSize
時候就會再次創建新的線程。
非核心線程 =(maximumPoolSize - corePoolSize ) ,非核心線程閑置下來不干活最多存活時間。
線程池中非核心線程保持存活的時間
TimeUnit.DAYS; 天 TimeUnit.HOURS; 小時 TimeUnit.MINUTES; 分鐘 TimeUnit.SECONDS; 秒 TimeUnit.MILLISECONDS; 毫秒 TimeUnit.MICROSECONDS; 微秒 TimeUnit.NANOSECONDS; 納秒
線程池 等待隊列,維護著等待執行的
Runnable
對象。當運行當線程數= corePoolSize時,新的任務會被添加到workQueue
中,如果workQueue
也滿了則嘗試用非核心線程執行任務,另外等待隊列盡量用有界的哦!!
創建一個新線程時使用的工廠,可以用來設定線程名、是否為daemon線程等等。
corePoolSize
、workQueue
、maximumPoolSize
都不可用的時候執行的 飽和策略。AbortPolicy :直接拋出異常,默認用此 CallerRunsPolicy:用調用者所在的線程來執行任務 DiscardOldestPolicy:丟棄阻塞隊列里最老的任務,隊列里最靠前的任務 DiscardPolicy :當前任務直接丟棄 想實現自己的飽和策略,實現RejectedExecutionHandler接口即可
形象流程圖如下:
// 核心思想跟上面的流程圖類似
public void execute(Runnable command) {
if (command == null) //規范性檢查
throw new NullPointerException();
int c = ctl.get();//當前工作的線程數跟線程狀態 ctl = AtomicInteger CAS級別
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
// 如果當前線程池中工作線程數小于核心線程數,直接添加任務 然后return
return;
c = ctl.get();// 添加失敗了重新獲得線程池中工作線程數
}
if (isRunning(c) && workQueue.offer(command)) {
// 線程池狀態是否處于可用,可用就嘗試將線程添加到queue
int recheck = ctl.get();// 獲得線程池狀態
if (! isRunning(recheck) && remove(command))
reject(command);// 如果線程狀態不在運行中 則remove 該任務
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
else if (!addWorker(command, false))// 嘗試將任務用非核心線程執行,
reject(command);//失敗了則執行失敗策略。
}
ThreadPoolExecutor extends AbstractExecutorService
父類中存在一個submit方法,
public <T> Future<T> submit(Callable<T> task) {
if (task == null) throw new NullPointerException();
RunnableFuture<T> ftask = newTaskFor(task);
execute(ftask);
return ftask;
}
注意線程之間是協作式的哦,所以的關閉只是發出關閉指令。
停止接收外部submit的任務 內部正在跑的任務和隊列里等待的任務,會執行完 等到第二步完成后,才真正停止
跟shutdown()一樣,先停止接收外部提交的任務 忽略隊列里等待的任務 嘗試將正在跑的任務interrupt中斷 返回未執行的任務列表
shutdown 跟shutdownnow簡單來說區別如下:
shutdownNow()能立即停止線程池,正在跑的和正在等待的任務都停下了。這樣做立即生效,但是風險也比較大。shutdown()只是關閉了提交通道,用submit()是無效的;而內部該怎么跑還是怎么跑,跑完再停。
pool.showdown()
boolean b = pool.awaitTermination(3, TimeUnit.SECONDS)
awaitTermination
有兩個參數,一個是timeout即超時時間,另一個是unit即時間單位。這個方法會使線程等待timeout時長,當超過timeout時間后,會監測ExecutorService
是否已經關閉,若關閉則返回true,否則返回false。一般情況下會和shutdown方法組合使用,調用后當前線程會阻塞,直到
等所有已提交的任務(包括正在跑的和隊列中等待的)執行完 或者等超時時間到 或者線程被中斷,拋出InterruptedException
總結
優雅的關閉,用shutdown() 想立馬關閉,并得到未執行任務列表,用shutdownNow() 優雅的關閉,發出關閉指令后看下是否真的關閉了用awaitTermination()。
線程在Java中屬于稀缺資源,線程池不是越大越好也不是越小越好。任務分為計算密集型、IO密集型、混合型。
計算密集型:大部分都在用CPU跟內存,加密,邏輯操作業務處理等。 IO密集型:數據庫鏈接,網絡通訊傳輸等。
Runtime.getRuntime().availableProcessors();
每個線程池都是一個實現了接口ExecutorService
并且繼承自ThreadPoolExecutor
的具體實現類,這些類的創建統一由一個工廠類Executors
來提供對外創建接口。Executors框架圖如下:
ThreadPoolExecutor
中一個線程就是一個Worker
對象,它與一個線程綁定,當Worker
執行完畢就是線程執行完畢。而Worker帶了鎖AQS,根據我后面準備寫的讀寫鎖的例子,發現線程池是線程安全的。看看圖二的類圖。下面簡單介紹幾個常用的線程池模式。
LinkedBlockingQueue
。使用時候小心堵滿等待隊列。
只有一條線程來執行任務,適用于有順序的任務的應用場景,也是用的無界等待隊列
可緩存的線程池,該線程池中沒有核心線程,非核心線程的數量為Integer.max_value,就是無限大,當有需要時創建線程來執行任務,沒有需要時回收線程,適用于耗時少,任務量大的情況。任務隊列用的是SynchronousQueue如果生產多快消費慢,則會導致創建很多線程需注意。
JDK7以后 基于ForkJoinPool實現。PS:其中FixedThreadPool
、SingleThreadPool
、CachedThreadPool
都用的無界等待隊列,因此實際工作中都不建議這樣做的哦,阿里巴巴Java編程規范建議如下:最后來個簡單的線程使用demo:
public class UseThreadPool
{
// 工作線程
static class Worker implements Runnable
{
private String taskName;
private Random r = new Random();
public Worker(String taskName)
{
this.taskName = taskName;
}
public String getName()
{
return taskName;
}
@Override
public void run()
{
System.out.println(Thread.currentThread().getName() + " 當前任務: " + taskName);
try
{
TimeUnit.MILLISECONDS.sleep(r.nextInt(100) * 5);
} catch (Exception e)
{
e.printStackTrace();
}
}
}
static class CallWorker implements Callable<String>
{
private String taskName;
private Random r = new Random();
public CallWorker(String taskName)
{
this.taskName = taskName;
}
public String getName()
{
return taskName;
}
@Override
public String call() throws Exception
{
System.out.println(Thread.currentThread().getName() + " 當前任務 : " + taskName);
return Thread.currentThread().getName() + ":" + r.nextInt(100) * 5;
}
}
public static void main(String[] args) throws InterruptedException, ExecutionException
{
ExecutorService pool = new ThreadPoolExecutor(2, 4, 3, TimeUnit.SECONDS,
new ArrayBlockingQueue<Runnable>(10),
new ThreadPoolExecutor.DiscardOldestPolicy());
// ExecutorService pool = Executors.newCachedThreadPool();
for (int i = 0; i < 5; i++)
{
Worker worker = new Worker("Runnable_" + i);
pool.execute(worker);
}
for (int i = 0; i < 5; i++)
{
CallWorker callWorker = new CallWorker("CallWorker_" + i);
Future<String> result = pool.submit(callWorker);
System.out.println(result.get());
}
pool.shutdown();
}
}
周期性執行任務的線程池,按照某種特定的計劃執行線程中的任務,有核心線程,但也有非核心線程,非核心線程的大小也為無限大。適用于執行周期性的任務。看構造函數:調用的還是ThreadPoolExecutor
構造函數,區別不同點在于任務隊列是用的DelayedWorkQueue,沒什么新奇的了。
核心函數講解:
FixedDelay
其中
scheduleAtFixedRate
跟
scheduleWithFixedDelay
區別如下圖
scheduleAtFixedRate任務超時狀態,比如我們設定60s執行一次,其中第一個任務時長 80s,第二個任務20s,第三個任務 50s。
第一個任務第0秒開始,第80s結束. 第二個任務第80s開始,在第100s結束. 第三個任務第120s秒開始,170s結束. 第四個任務從180s開始.
簡單Mark個循環任務demo:
class ScheduleWorker implements Runnable {
public final static int Normal = 0;//普通任務類型
public final static int HasException = -1;//會拋出異常的任務類型
public final static int ProcessException = 1;//拋出異常但會捕捉的任務類型
public static SimpleDateFormat formater = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
private int taskType;
public ScheduleWorker(int taskType) {
this.taskType = taskType;
}
@Override
public void run() {
if (taskType == HasException) {
System.out.println(formater.format(new Date()) + " 異常產生");
throw new RuntimeException("有異常");
} else if (taskType == ProcessException) {
try {
System.out.println(formater.format(new Date()) + " 異常產生被捕捉");
throw new RuntimeException("異常被捕捉");//異常導致下個任務無法執行
} catch (Exception e) {
System.out.println(" 異常被主播");
}
} else {
System.out.println("正常" + formater.format(new Date()));
}
}
}
public class SchTest{
public static void main(String[] args) {
ScheduledThreadPoolExecutor schedule = new ScheduledThreadPoolExecutor(1);
schedule.scheduleAtFixedRate(new ScheduleWorker(ScheduleWorker.HasException),
1000, 3000, TimeUnit.MILLISECONDS); // 任務在 1秒后執行 周期3秒
schedule.scheduleAtFixedRate(new ScheduleWorker(ScheduleWorker.Normal),
1000, 3000, TimeUnit.MILLISECONDS);
}
}
JDK8中新添加的一個類,攝像一個場景你去詢問兩個商品價格然后將價格保存數據庫。
ExecutorService executor =Executors.newFixedThreadPool(2);
// 異步向電商 S1 詢價
Future<Integer> f1 = executor.submit(()->getPriceByS1());
// 異步向電商 S2 詢價
Future<Integer> f2 = executor.submit(()->getPriceByS2());
// 獲取電商 S1 報價并保存
r=f1.get();
executor.execute(()->save(r));
// 獲取電商 S2 報價并保存
r=f2.get();
executor.execute(()->save(r));
上面的這個方案本身沒有太大問題,但是有個地方的處理需要你注意,那就是如果獲取電商 S1 報價的耗時很長,那么即便獲取電商 S2 報價的耗時很短,也無法讓保存 S2 報價的操作先執行,因為這個主線程都阻塞在了 f1.get(),那我們如何解決了?解決方法:結果都存入到一個阻塞隊列中去。
// 創建阻塞隊列
BlockingQueue<Integer> bq =new LinkedBlockingQueue<>();
// 電商 S1 報價異步進入阻塞隊列
executor.execute(()->bq.put(f1.get()));
// 電商 S2 報價異步進入阻塞隊列
executor.execute(()->bq.put(f2.get()));
// 異步保存所有報價
for (int i=0; i<2; i++) {
Integer r = bq.take();
executor.execute(()->save(r));
}
在JDK8中不建議上面的工作都手動實現,JDK提供了CompletionService
,它實現原理也是內部維護了一個阻塞隊列,它的核心功效就是讓先執行的任務先放到結果集。當任務執行結束就把任務的執行結果加入到阻塞隊列中,不同的是CompletionService
是把任務執行結果的 Future 對象加入到阻塞隊列中,而上面的示例代碼是把任務最終的執行結果放入了阻塞隊列中。CompletionService
將Executor
和BlockingQueue
的功能融合在一起,CompletionService
內部有個阻塞隊列。CompletionService
接口的實現類是 ExecutorCompletionService
,這個實現類的構造方法有兩個,分別是:
ExecutorCompletionService(Executor executor)
ExecutorCompletionService(Executor executor, BlockingQueue<Future<V>> completionQueue)
這兩個構造方法都需要傳入一個線程池,如果不指定 completionQueue,那么默認會使用無界的 LinkedBlockingQueue。任務執行結果的 Future 對象就是加入到 completionQueue 中。
// 創建線程池
ExecutorService executor = Executors.newFixedThreadPool(2);
// 創建 CompletionService
CompletionService<Integer> cs = new ExecutorCompletionService<>(executor);
// 異步向電商 S1 詢價
cs.submit(()->getPriceByS1());
// 異步向電商 S2 詢價
cs.submit(()->getPriceByS2());
// 將詢價結果異步保存到數據庫
for (int i=0; i<2; i++) {
Integer r = cs.take().get();
executor.execute(()->save(r));
}
來一個整體的demo加深印象:
// 任務類
class WorkTask implements Callable<Integer>
{
private String name;
public WorkTask(String name)
{
this.name = name;
}
@Override
public Integer call()
{
int sleepTime = new Random().nextInt(1000);
try
{
Thread.sleep(sleepTime);
} catch (InterruptedException e)
{
e.printStackTrace();
}
return sleepTime;
}
}
public class CompletionCase
{
private final int POOL_SIZE = Runtime.getRuntime().availableProcessors();
private final int TOTAL_TASK = Runtime.getRuntime().availableProcessors();
public void selfByQueue() throws Exception
{
long start = System.currentTimeMillis(); // 統計所有任務休眠的總時長
AtomicInteger count = new AtomicInteger(0);
ExecutorService pool = Executors.newFixedThreadPool(POOL_SIZE); // 創建線程池
BlockingQueue<Future<Integer>> queue = new LinkedBlockingQueue<Future<Integer>>();//容器存放提交給線程池的任務,list,map,
for (int i = 0; i < TOTAL_TASK; i++)
{
Future<Integer> future = pool.submit(new WorkTask("要執行的第幾個任務" + i));
queue.add(future);//i=0 先進隊列,i=1的任務跟著進
}
for (int i = 0; i < TOTAL_TASK; i++)
{
int sleptTime = queue.take().get(); // 檢查線程池任務執行結果 i=0先取到,i=1的后取到
System.out.println(" 休眠毫秒數 = " + sleptTime + " ms ");
count.addAndGet(sleptTime);
}
pool.shutdown();
System.out.println("休眠時間" + count.get() + "ms,耗時時間" + (System.currentTimeMillis() - start) + " ms");
}
public void testByCompletion() throws Exception
{
long start = System.currentTimeMillis();
AtomicInteger count = new AtomicInteger(0);
// 創建線程池
ExecutorService pool = Executors.newFixedThreadPool(POOL_SIZE);
CompletionService<Integer> cService = new ExecutorCompletionService<>(pool);
// 向里面扔任務
for (int i = 0; i < TOTAL_TASK; i++)
{
cService.submit(new WorkTask("執行任務" + i));
}
// 檢查線程池任務執行結果
for (int i = 0; i < TOTAL_TASK; i++)
{
int sleptTime = cService.take().get();
System.out.println("休眠毫秒數 = " + sleptTime + " ms ...");
count.addAndGet(sleptTime);
}
pool.shutdown();
System.out.println("休眠時間 " + count.get() + "ms,耗時時間" + (System.currentTimeMillis() - start) + " ms");
}
public static void main(String[] args) throws Exception
{
CompletionCase t = new CompletionCase();
t.selfByQueue();
t.testByCompletion();
}
}
為什么用線程池? 線程池的作用? 常用的線程池模版? 7大重要參數? 4大拒絕策略? 常見線程池任務隊列,如何理解有界跟無界?7.如何分配線程池個數? 單機線程池執行一般斷電了如何考慮? 正在處理的實現事務功能,下次自動回滾,隊列實現持久化儲存,下次啟動自動載入。
設定一個線程池優先級隊列,Runable類要實現可對比功能,任務隊列使用優先級隊列
到此,關于“怎么快速了解Java線程池”的學習就結束了,希望能夠解決大家的疑惑。理論與實踐的搭配能更好的幫助大家學習,快去試試吧!若想繼續學習更多相關知識,請繼續關注億速云網站,小編會繼續努力為大家帶來更多實用的文章!
免責聲明:本站發布的內容(圖片、視頻和文字)以原創、轉載和分享為主,文章觀點不代表本網站立場,如果涉及侵權請聯系站長郵箱:is@yisu.com進行舉報,并提供相關證據,一經查實,將立刻刪除涉嫌侵權內容。