執行緒池應用及實現原理剖析
阿新 • • 發佈：2019-12-31
為什麼要用執行緒池
執行緒是不是越多好？
- 執行緒在Java中是一個物件， 更是作業系統的資源， 執行緒額建立和銷燬都需要時間，如果 建立時間+銷燬時間>執行任務時間 就很不合算
- Java物件佔用堆記憶體，作業系統執行緒佔用系統記憶體， 根據JVM規範，一個執行緒預設最大棧大小為1M， 這個棧空間是要從作業系統記憶體中分配的，執行緒過多會消耗很多的記憶體
- 作業系統頻繁切換執行緒上下文會影響效能
執行緒池的推出就是為了控制執行緒數量
執行緒池原理 - 概念
- 執行緒池管理器：用於建立並管理執行緒池， 包括建立執行緒池， 銷燬執行緒池， 新增新任務
- 工作執行緒： 執行緒池中的執行緒， 在沒有任務時處於等待狀態， 可以迴圈的執行任務
- 任務介面： 每個任務必須實現的介面，以供工作執行緒任務排程的執行， 它主要規定了任務的入口，任務執行完後的收尾工作，任務的執行狀態等
- 任務佇列： 用於存放沒有處理的任務，提供一種緩衝
執行緒池API - 介面定義和實現類
執行緒池API - 方法定義
ExecutorService
ScheduledExecutorService
執行緒池API - Executors工具類
可以自己例項化執行緒池， 也可以用Executors建立執行緒池，方法如下：
- newFixedThreadPool(int nThreads) : 建立一個固定大小，任務佇列無界的的執行緒池，核心執行緒數=最大執行緒數
- newCachedThreadPool() : 建立一個大小無界的緩衝執行緒池，它的任務佇列是一個同步佇列，任務加入加入到池中，如果池中有空閒執行緒， 則用空閒執行緒執行，如無則建立新執行緒執行，池中的空閒執行緒超過60秒， 將被銷燬釋放， 執行緒數隨任務的多少變化，適用於耗時較小的非同步任務， 池的核心執行緒數=0， 最大執行緒數=Integer.MAX_VALUE
- newSingleThreadExecutor() : 只有一個執行緒來執行無界任務佇列的單一執行緒池，該執行緒池確保任務按加入的順序一個一個的依次執行，當唯一的執行緒因任務異常中止時，將建立一個新的執行緒來繼續執行後續的任務， 與newFixedThreadPool(1)的區別在於，單一執行緒池的池大小在newSingleThreadExecutor方法中硬編碼，不能再改變
- newScheduledThreadPool(int corePoolSize) : 能定時執行任務的執行緒池，該池的核心執行緒數由引數指定，最大執行緒數=Integer.MAX_VALUE
執行緒池原理 - 任務執行過程
- 是否達到核心執行緒數量？ 沒達到，建立一個工作執行緒來執行任務
- 工作佇列是否已滿？沒滿，則將新提交的任務儲存在任務佇列中
- 是否達到執行緒池最大數量？則建立一個新的執行緒來執行任務
- 最後，執行拒絕策略來處理這個任務
程式碼示例：
package com.neteasy.demo;
import java.util.List;
import java.util.concurrent.*;
/** 執行緒池的使用 */
public class Demo7 {
/**
* 測試： 提交15個執行時間需要3秒的任務,看執行緒池的狀況
*
* @param threadPoolExecutor
* @throws Exception
*/
private void testCommon(ThreadPoolExecutor threadPoolExecutor) throws Exception {
// 測試： 提交15個執行時間需要3秒的任務，看超過大小的2個，對應的處理情況
for (int i = 0; i < 15; i++) {
int n = i;
threadPoolExecutor.submit(
new Runnable() {
@Override
public void run() {
try {
System.out.println("開始執行： " + n);
Thread.sleep(3000L);
System.out.println("執行結束： " + n);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
System.out.println("任務提交成功：" + i);
}
// 檢視執行緒數量，檢視佇列等待數量
Thread.sleep(500L);
System.out.println("當前執行緒池執行緒數量為：" + threadPoolExecutor.getPoolSize());
System.out.println("當前執行緒池等待的數量為" + threadPoolExecutor.getQueue().size());
// 等待15秒，檢視執行緒數量和佇列數量（理論上，會被超出核心執行緒數量的執行緒自動銷燬）
Thread.sleep(15000L);
System.out.println("當前執行緒池執行緒數量為：" + threadPoolExecutor.getPoolSize());
System.out.println("當前執行緒池等待的數量為：" + threadPoolExecutor.getQueue().size());
}
/**
* 1、執行緒池資訊： 核心執行緒數量5，最大數量10，無界佇列，超出核心執行緒數量的執行緒存活時間：5秒， 指定拒絕策略的
*
* @throws Exception
*/
private void threadPoolExecutorTest1() throws Exception {
ThreadPoolExecutor threadPoolExecutor =
new ThreadPoolExecutor(5,10,5,TimeUnit.SECONDS,new LinkedBlockingDeque<Runnable>());
testCommon(threadPoolExecutor);
// 預計結果：執行緒池執行緒數量為：5,超出數量的任務，其他的進入佇列中等待被執行
}
/**
* 2、 執行緒池資訊： 核心執行緒數量5，最大數量10，佇列大小3，超出核心執行緒數量的執行緒存活時間：5秒， 指定拒絕策略的
*
* @throws Exception
*/
private void threadPoolExecutorTest2() throws Exception {
// 建立一個 核心執行緒數量為5，最大數量為10,等待佇列最大是3 的執行緒池，也就是最大容納13個任務。
// 預設的策略是丟擲RejectedExecutionException異常，java.util.concurrent.ThreadPoolExecutor.AbortPolicy
ThreadPoolExecutor threadPoolExecutor =
new ThreadPoolExecutor(
5,new LinkedBlockingDeque<Runnable>(3),new RejectedExecutionHandler() {
@Override
public void rejectedExecution(Runnable r,ThreadPoolExecutor executor) {
System.err.println("有任務被拒絕執行了");
}
});
testCommon(threadPoolExecutor);
// 預計結果：
// 1、 5個任務直接分配執行緒開始執行
// 2、 3個任務進入等待佇列
// 3、 佇列不夠用，臨時加開5個執行緒來執行任務(5秒沒活幹就銷燬)
// 4、 佇列和執行緒池都滿了，剩下2個任務，沒資源了，被拒絕執行。
// 5、 任務執行，5秒後，如果無任務可執行，銷燬臨時建立的5個執行緒
}
/**
* 3、 執行緒池資訊： 核心執行緒數量5，最大數量5，無界佇列，超出核心執行緒數量的執行緒存活時間：5秒
*
* @throws Exception
*/
private void threadPoolExecutorTest3() throws Exception {
// 和Executors.newFixedThreadPool(int nThreads)一樣的
ThreadPoolExecutor threadPoolExecutor =
new ThreadPoolExecutor(
5,0L,TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>());
testCommon(threadPoolExecutor);
// 預計結：執行緒池執行緒數量為：5，超出數量的任務，其他的進入佇列中等待被執行
}
/**
* 4、 執行緒池資訊： 核心執行緒數量0，最大數量Integer.MAX_VALUE，SynchronousQueue佇列，超出核心執行緒數量的執行緒存活時間：60秒
*
* @throws Exception
*/
private void threadPoolExecutorTest4() throws Exception {
/**
* SynchronousQueue，實際上它不是一個真正的佇列，因為它不會為佇列中元素維護儲存空間。 與其他佇列不同的是，它維護一組執行緒，這些執行緒在等待著把元素加入或移出佇列。
* 在使用SynchronousQueue作為工作佇列的前提下，客戶端程式碼向執行緒池提交任務時， 而執行緒池中又沒有空閒的執行緒能夠從SynchronousQueue佇列例項中取一個任務，
* 那麼相應的offer方法呼叫就會失敗（即任務沒有被存入工作佇列）。 此時，ThreadPoolExecutor會新建一個新的工作者執行緒用於對這個入佇列失敗的任務進行處理
* （假設此時執行緒池的大小還未達到其最大執行緒池大小maximumPoolSize）。
*/
// 和Executors.newCachedThreadPool()一樣的
ThreadPoolExecutor threadPoolExecutor =
new ThreadPoolExecutor(
0,Integer.MAX_VALUE,60L,new SynchronousQueue<Runnable>());
testCommon(threadPoolExecutor);
// 預計結果：
// 1、 執行緒池執行緒數量為：15，超出數量的任務，其他的進入佇列中等待被執行
// 2、 所有任務執行結束，60秒後，如果無任務可執行，所有執行緒全部被銷燬，池的大小恢復為0
Thread.sleep(60000L);
System.out.println("60秒後，再看執行緒池中的數量：" + threadPoolExecutor.getPoolSize());
}
/**
* 5、 定時執行執行緒池資訊：3秒後執行，一次性任務，到點就執行
* 核心執行緒數量5，最大數量Integer.MAX_VALUE，DelayedWorkQueue延時佇列，超出核心執行緒數量的執行緒存活時間：0秒
*
* @throws Exception
*/
private void threadPoolExecutorTest5() throws Exception {
// 和Executors.newScheduledThreadPool()一樣的
ScheduledThreadPoolExecutor threadPoolExecutor = new ScheduledThreadPoolExecutor(5);
threadPoolExecutor.schedule(
new Runnable() {
@Override
public void run() {
System.out.println("任務被執行，現在時間：" + System.currentTimeMillis());
}
},3000,TimeUnit.MILLISECONDS);
System.out.println(
"定時任務，提交成功，時間是："
+ System.currentTimeMillis()
+ ",當前執行緒池中執行緒數量："
+ threadPoolExecutor.getPoolSize());
// 預計結果：任務在3秒後被執行一次
}
/**
* 6、 定時執行執行緒池資訊：執行緒固定數量5 ，<br>
* 核心執行緒數量5，最大數量Integer.MAX_VALUE，DelayedWorkQueue延時佇列，超出核心執行緒數量的執行緒存活時間：0秒
*
* @throws Exception
*/
private void threadPoolExecutorTest6() throws Exception {
ScheduledThreadPoolExecutor threadPoolExecutor = new ScheduledThreadPoolExecutor(5);
/**
* 週期性執行某一個任務，執行緒池提供了兩種排程方式，這裡單獨演示一下。測試場景一樣。 測試場景：提交的任務需要3秒才能執行完畢。看兩種不同排程方式的區別 效果1：
* 提交後，2秒後開始第一次執行，之後每間隔1秒，固定執行一次(如果發現上次執行還未完畢，則等待完畢，完畢後立刻執行)。
* 也就是說這個程式碼中是，3秒鐘執行一次（計算方式：每次執行三秒，間隔時間1秒，執行結束後馬上開始下一次執行，無需等待）
*/
threadPoolExecutor.scheduleAtFixedRate(
new Runnable() {
@Override
public void run() {
try {
Thread.sleep(3000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("任務-1 被執行，現在時間：" + System.currentTimeMillis());
}
},2000,1000,TimeUnit.MILLISECONDS);
/**
* 效果2：提交後，2秒後開始第一次執行，之後每間隔1秒，固定執行一次(如果發現上次執行還未完畢，則等待完畢，等上一次執行完畢後再開始計時，等待1秒)。
* 也就是說這個程式碼鐘的效果看到的是：4秒執行一次。 （計算方式：每次執行3秒，間隔時間1秒，執行完以後再等待1秒，所以是 3+1）
*/
threadPoolExecutor.scheduleWithFixedDelay(
new Runnable() {
@Override
public void run() {
try {
Thread.sleep(3000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("任務-2 被執行，現在時間：" + System.currentTimeMillis());
}
},TimeUnit.MILLISECONDS);
}
private ThreadPoolExecutor testShutdownCommon() {
// 建立一個 核心執行緒數量為5，最大數量為10,等待佇列最大是3的執行緒池，也就是最大容納13個任務。
// 預設的策略是丟擲RejectedExecutionException異常，java.util.concurrent.ThreadPoolExecutor.AbortPolicy
ThreadPoolExecutor threadPoolExecutor =
new ThreadPoolExecutor(
5,new LinkedBlockingQueue<Runnable>(3),ThreadPoolExecutor executor) {
System.err.println("有任務被拒絕執行了");
}
});
// 測試： 提交15個執行時間需要3秒的任務，看超過大小的2個，對應的處理情況
for (int i = 0; i < 15; i++) {
int n = i;
threadPoolExecutor.submit(
new Runnable() {
@Override
public void run() {
try {
System.out.println("開始執行：" + n);
Thread.sleep(3000L);
System.err.println("執行結束:" + n);
} catch (InterruptedException e) {
System.out.println("異常：" + e.getMessage());
}
}
});
System.out.println("任務提交成功 :" + i);
}
return threadPoolExecutor;
}
/**
* 7、 終止執行緒：執行緒池資訊： 核心執行緒數量5，最大數量10，佇列大小3，超出核心執行緒數量的執行緒存活時間：5秒， 指定拒絕策略的
*
* @throws Exception
*/
private void threadPoolExecutorTest7() throws Exception {
ThreadPoolExecutor threadPoolExecutor = testShutdownCommon();
// 1秒後終止執行緒池
Thread.sleep(1000L);
threadPoolExecutor.shutdown();
// 再次提交提示失敗
threadPoolExecutor.submit(
new Runnable() {
@Override
public void run() {
System.out.println("追加一個任務");
}
});
// 結果分析
// 1、 10個任務被執行，3個任務進入佇列等待，2個任務被拒絕執行
// 2、呼叫shutdown後，不接收新的任務，等待13任務執行結束
// 3、 追加的任務線上程池關閉後，無法再提交，會被拒絕執行
}
/**
* 8、 立刻終止執行緒：執行緒池資訊： 核心執行緒數量5，最大數量10，佇列大小3，超出核心執行緒數量的執行緒存活時間：5秒， 指定拒絕策略的
*
* @throws Exception
*/
private void threadPoolExecutorTest8() throws Exception {
// 建立一個 核心執行緒數量為5，最大數量為10,等待佇列最大是3 的執行緒池，也就是最大容納13個任務。
// 預設的策略是丟擲RejectedExecutionException異常，java.util.concurrent.ThreadPoolExecutor.AbortPolicy
ThreadPoolExecutor threadPoolExecutor = testShutdownCommon();
// 1秒後終止執行緒池
Thread.sleep(1000L);
List<Runnable> shutdownNow = threadPoolExecutor.shutdownNow();
// 再次提交提示失敗
threadPoolExecutor.submit(
new Runnable() {
@Override
public void run() {
System.out.println("追加一個任務");
}
});
System.out.println("未結束的任務有：" + shutdownNow.size());
// 結果分析
// 1、 10個任務被執行，3個任務進入佇列等待，2個任務被拒絕執行
// 2、呼叫shutdownnow後，佇列中的3個執行緒不再執行，10個執行緒被終止
// 3、 追加的任務線上程池關閉後，無法再提交，會被拒絕執行
}
public static void main(String[] args) throws Exception {
Demo7 demo7 = new Demo7();
// demo7.threadPoolExecutorTest1();
// demo7.threadPoolExecutorTest2();
// demo7.threadPoolExecutorTest3();
// demo7.threadPoolExecutorTest4();
// demo7.threadPoolExecutorTest5();
// demo7.threadPoolExecutorTest6();
// demo7.threadPoolExecutorTest7();
demo7.threadPoolExecutorTest8();
}
}
複製程式碼