java多線程系列:ThreadPoolExecutor

NO IMAGE

ThreadPoolExecutor自定義線程池

開篇一張圖(圖片來自阿里巴巴Java開發手冊(詳盡版)),後面全靠編

java多線程系列:ThreadPoolExecutor

好了要開始編了,從圖片中就可以看到這篇博文的主題了,ThreadPoolExecutor自定義線程池。

目錄

  1. ThreadPoolExecutor構造函數介紹
  2. 核心線程數corePoolSize
  3. 最大線程數maximumPoolSize
  4. 線程存活時間keepAliveTime
  5. 線程存活時間單位unit
  6. 創建線程的工廠threadFactory
  7. 隊列
  8. 拒絕策略
  9. 線程池擴展

ThreadPoolExecutor構造函數介紹

在介紹穿件線程池的方法之前要先介紹一個類ThreadPoolExecutor,因為Executors工廠大部分方法都是返回ThreadPoolExecutor對象,先來看看它的構造函數吧

public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {...}

參數介紹

參數類型含義
corePoolSizeint核心線程數
maximumPoolSizeint最大線程數
keepAliveTimelong存活時間
unitTimeUnit時間單位
workQueueBlockingQueue存放線程的隊列
threadFactoryThreadFactory創建線程的工廠
handlerRejectedExecutionHandler多餘的的線程處理器(拒絕策略)

核心線程數corePoolSize

這個參數表示線程池中的基本線程數量也就是核心線程數量。

最大線程數maximumPoolSize

這個參數是線程池中允許創建的最大線程數量,當使用有界隊列時,且隊列存放的任務滿了,那麼線程池會創建新的線程(最大不會超過這個參數所設置的值)。需要注意的是,當使用無界隊列時,這個參數是無效的。

線程存活時間keepAliveTime

這個就是線程空閒時可以存活的時間,一旦超過這個時間,線程就會被銷燬。

線程存活時間單位unit

線程存活的時間單位,有NANOSECONDS(納秒)、MICROSECONDS(微秒)、MILLISECONDS(毫秒)、SECONDS(秒)、MINUTES(分鐘)、HOURS(小時)、DAYS(天)。TimeUnit代碼如下

public enum TimeUnit {
NANOSECONDS {...},
MICROSECONDS {...},
MILLISECONDS {...},
SECONDS {...},
MINUTES {...},
HOURS {...},
DAYS {...};
}

創建線程的工廠threadFactory

創建線程的工廠,一般都是採用Executors.defaultThreadFactory()方法返回的DefaultThreadFactory,當然也可以用其他的來設置更有意義的名稱。

DefaultThreadFactory類如下

/**
* The default thread factory
*/
static class DefaultThreadFactory implements ThreadFactory {
private static final AtomicInteger poolNumber = new AtomicInteger(1);
private final ThreadGroup group;
private final AtomicInteger threadNumber = new AtomicInteger(1);
private final String namePrefix;
DefaultThreadFactory() {
SecurityManager s = System.getSecurityManager();
group = (s != null) ? s.getThreadGroup() :
Thread.currentThread().getThreadGroup();
namePrefix = "pool-" +
poolNumber.getAndIncrement() +
"-thread-";
}
public Thread newThread(Runnable r) {
Thread t = new Thread(group, r,
namePrefix + threadNumber.getAndIncrement(),
0);
if (t.isDaemon())
t.setDaemon(false);
if (t.getPriority() != Thread.NORM_PRIORITY)
t.setPriority(Thread.NORM_PRIORITY);
return t;
}
}

隊列

分為有界隊列和無界隊列,用於存放等待執行的任務的阻塞隊列。有SynchronousQueue、ArrayBlockingQueue、LinkedBlockingQueue、DelayQueue、PriorityBlockingQueue、LinkedTransferQueue、DelayedWorkQueue、LinkedBlockingDeque。下面將介紹有界和無界兩種常用的隊列。BlockingQueue類圖如下

java多線程系列:ThreadPoolExecutor

有界隊列

當使用有界隊列時,如果有新的任務需要添加進來時,如果線程池實際線程數小於corePoolSize(核心線程數),則優先創建線程,如果線程池實際線程數大於corePoolSize(核心線程數),則會將任務加入隊列,若隊列已滿,則在中現場數不大於maximumPoolSize(最大線程數)的前提下,創建新的線程,若線程數大於maximumPoolSize(最大線程數),則執行拒絕策略。

無界隊列

當使用無界隊列時,maximumPoolSize(最大線程數)和拒絕策略便會失效,因為隊列是沒有限制的,所以就不存在隊列滿的情況。和有界隊列相比,當有新的任務添加進來時,都會進入隊列等待。但是這也會出現一些問題,例如線程的執行速度比任務提交速度慢,會導致無界隊列快速增長,直到系統資源耗盡。

拒絕策略

當使用有界隊列時,且隊列任務被填滿後,線程數也達到最大值時,拒絕策略開始發揮作用。ThreadPoolExecutor默認使用AbortPolicy拒絕策略。RejectedExecutionHandler類圖如下

java多線程系列:ThreadPoolExecutor

我們來看看ThreadPoolExecutor是如何調用RejectedExecutionHandler的,可以直接查看execute方法

public class ThreadPoolExecutor extends AbstractExecutorService {
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
int c = ctl.get();
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
c = ctl.get();
}
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
if (! isRunning(recheck) && remove(command))
reject(command);
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}else if (!addWorker(command, false))
//拒絕線程
reject(command);
}
}

可以看到經過一系列的操作,不符合條件的會調用reject方法,那我麼接著來看看reject方法

final void reject(Runnable command) {
handler.rejectedExecution(command, this);
}

可以看到調用了RejectedExecutionHandler接口的rejectedExecution方法。好了,現在來看看jdk提供的幾個拒絕策略。

拒絕策略的測試代碼在這

注:後續會寫一篇ThreadPoolExecutor源碼解析,專門介紹ThreadPoolExecutor各個流程

AbortPolicy

從下面代碼可以看到直接拋出異常信息,但是線程池還是可以正常工作的。

public static class AbortPolicy implements RejectedExecutionHandler {
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
throw new RejectedExecutionException("Task " + r.toString() +
" rejected from " +
e.toString());
}
}

示例代碼

線程類

public class Task implements Runnable{
private int id ;
public Task(int id){
this.id = id;
}
public int getId() {
return id;
}
public void setId(int id) {
this.id = id;
}
@Override
public void run() {
//
System.out.println(LocalTime.now()+" 當前線程id和名稱為:" + this.id);
try {
Thread.sleep(1000);
} catch (Exception e) {
e.printStackTrace();
}
}
public String toString(){
return "當前線程的內容為:{ id : " + this.id + "}";
}
}

測試代碼

public class TestAbortPolicy {
public static void main(String[] args) {
//定義了1個核心線程數,最大線程數1個,隊列長度2個
ThreadPoolExecutor executor = new ThreadPoolExecutor(
1,
1,
60,
TimeUnit.SECONDS,
new ArrayBlockingQueue<Runnable>(2),
new ThreadPoolExecutor.AbortPolicy());
//直接提交4個線程
executor.submit(new Task(1));
executor.submit(new Task(2));
executor.submit(new Task(3));
//提交第四個拋異常
executor.submit(new Task(4));
}
}

執行結果

當前線程id和名稱為:1
Exception in thread "main" java.util.concurrent.RejectedExecutionException: Task [email protected] rejected from [email protected][Running, pool size = 1, active threads = 1, queued tasks = 2, completed tasks = 0]
at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2047)
at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:823)
at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1369)
at java.util.concurrent.AbstractExecutorService.submit(AbstractExecutorService.java:112)
at com.learnConcurrency.executor.customThreadPool.testRejectedExecutionHandler.TestAbortPolicy.main(TestAbortPolicy.java:25)
當前線程id和名稱為:2
當前線程id和名稱為:3

可以看到添加第四個線程是拋出異常

CallerRunsPolicy

首先判斷線程池是否關閉,如果未關閉,則直接執行該線程。關閉則不做任何事情。

public static class CallerRunsPolicy implements RejectedExecutionHandler {
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
r.run();
}
}
}

代碼和上面的差不多就不貼了,想要查看的可以到github上查看TestCallerRunsPolicy,執行結果如下

14:58:19.462 當前線程id和名稱為:4
14:58:19.462 當前線程id和名稱為:1
14:58:20.464 當前線程id和名稱為:5
14:58:20.464 當前線程id和名稱為:2
14:58:21.464 當前線程id和名稱為:3
14:58:22.464 當前線程id和名稱為:6

DiscardPolicy

可以看到裡面沒有任何代碼,也就是這個被拒絕的線程任務被丟棄了,不作任何處理。

public static class DiscardPolicy implements RejectedExecutionHandler {
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
}
}

DiscardOldestPolicy

首先判斷線程池是否關閉,如果未關閉,丟棄最老的一個請求,嘗試再次提交當前任務。 關閉則不做任何事情。

public static class DiscardOldestPolicy implements RejectedExecutionHandler {
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
e.getQueue().poll();
e.execute(r);
}
}
}

代碼和上面的差不多就不貼了,想要查看的可以到github上查看TestDiscardOldestPolicy,執行結果如下

15:02:28.484 當前線程id和名稱為:1
15:02:29.486 當前線程id和名稱為:5
15:02:30.487 當前線程id和名稱為:6

可以看到線程2、3、4都被替換了

自定義拒絕策略

實現RejectedExecutionHandle接口即可,如下MyRejected

public class MyRejected implements RejectedExecutionHandler{
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
System.out.println("自定義處理:開始記錄日誌");
System.out.println(r.toString());
System.out.println("自定義處理:記錄日誌完成");
}
}

測試代碼

public class TestCustomeRejectedPolicy {
public static void main(String[] args) {
//定義了1個核心線程數,最大線程數1個,隊列長度2個
ThreadPoolExecutor executor = new ThreadPoolExecutor(
1,
1,
60,
TimeUnit.SECONDS,
new ArrayBlockingQueue<Runnable>(2),
new MyRejected());
executor.execute(new Task(1));
executor.execute(new Task(2));
executor.execute(new Task(3));
executor.execute(new Task(4));
executor.execute(new Task(5));
executor.execute(new Task(6));
executor.shutdown();
}
}

輸出結果

自定義處理:開始記錄日誌
當前線程的內容為:{ id : 4}
自定義處理:記錄日誌完成
自定義處理:開始記錄日誌
當前線程的內容為:{ id : 5}
自定義處理:記錄日誌完成
自定義處理:開始記錄日誌
當前線程的內容為:{ id : 6}
自定義處理:記錄日誌完成
15:12:39.267 當前線程id和名稱為:1
15:12:40.268 當前線程id和名稱為:2
15:12:41.268 當前線程id和名稱為:3
Process finished with exit code 0

這裡如果有仔細觀察的你可能會有所好奇,為什麼這裡用execute方法而不是用submit?

這時因為用submit方法後,傳入的線程會被封裝成RunnableFuture,而我寫的MyRejected有調用到toString方法,Task類有重寫toString方法,但是被封裝成RunnableFuture會輸入如下內容

自定義處理:開始記錄日誌
[email protected]
自定義處理:記錄日誌完成
自定義處理:開始記錄日誌
[email protected]
自定義處理:記錄日誌完成
自定義處理:開始記錄日誌
[email protected]
自定義處理:記錄日誌完成
15:18:17.262 當前線程id和名稱為:1
15:18:18.263 當前線程id和名稱為:2
15:18:19.264 當前線程id和名稱為:3
Process finished with exit code 0

線程池擴展

ThreadPoolExecutor類中有三個方法是空方法,可以通過繼承來重寫這三個方法對線程進行監控。通過重寫beforeExecute和afterExecute方法,可以添加日誌、計時、監控等等功能。terminated方法是在線程關閉時調用的,可以在這裡面進行通知、日誌等操作。

//任務執行前
protected void beforeExecute(Thread t, Runnable r) { }
//任務執行後
protected void afterExecute(Runnable r, Throwable t) { }
//線程池關閉
protected void terminated() { }

示例代碼

public class Main {
public static void main(String[] args) {
ThreadPoolExecutor pool = new MyThreadPoolExecutor(
2,              //coreSize
4,              //MaxSize
60,          //60
TimeUnit.SECONDS,
new ArrayBlockingQueue<Runnable>(4));
for (int i = 0; i < 8; i++) {
int finalI = i + 1;
pool.submit(() -> {
try {
Thread.sleep(new Random().nextInt(1000));
} catch (InterruptedException e) {
e.printStackTrace();
}
});
}
pool.shutdown();
}
static class MyThreadPoolExecutor extends ThreadPoolExecutor{
private final AtomicInteger tastNum = new AtomicInteger();
private final ThreadLocal<Long> startTime = new ThreadLocal<>();
public MyThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
}
@Override
protected void beforeExecute(Thread t, Runnable r) {
super.beforeExecute(t, r);
startTime.set(System.nanoTime());
System.out.println(LocalTime.now()+" 執行之前-任務:"+r.toString());
}
@Override
protected void afterExecute(Runnable r, Throwable t) {
long endTime = System.nanoTime();
long time = endTime - startTime.get();
tastNum.incrementAndGet();
System.out.println(LocalTime.now()+" 執行之後-任務:"+r.toString()+",花費時間(納秒):"+time);
super.afterExecute(r, t);
}
@Override
protected void terminated() {
System.out.println("線程關閉,總共執行線程數:"+tastNum.get());
super.terminated();
}
}
}

執行結果

15:43:23.329 執行之前-任務:[email protected]
15:43:23.329 執行之前-任務:[email protected]
15:43:23.329 執行之前-任務:[email protected]
15:43:23.329 執行之前-任務:[email protected]
15:43:23.513 執行之後-任務:[email protected],花費時間(納秒):216399556
15:43:23.513 執行之前-任務:[email protected]
15:43:23.601 執行之後-任務:[email protected],花費時間(納秒):304505594
15:43:23.601 執行之前-任務:[email protected]
15:43:23.733 執行之後-任務:[email protected],花費時間(納秒):436283680
15:43:23.733 執行之前-任務:[email protected]
15:43:23.808 執行之後-任務:[email protected],花費時間(納秒):512242583
15:43:23.808 執行之前-任務:[email protected]
15:43:23.924 執行之後-任務:[email protected],花費時間(納秒):322900976
15:43:24.059 執行之後-任務:[email protected],花費時間(納秒):546324680
15:43:24.498 執行之後-任務:[email protected],花費時間(納秒):765309335
15:43:24.594 執行之後-任務:[email protected],花費時間(納秒):785868205
線程關閉,總共執行線程數:8

代碼位置

GitHub地址

地址在這

覺得不錯的點個star

參考資料

[1] Java 併發編程的藝術

[2] Java 併發編程實戰

相關文章

redis系列:通過demo學習hash命令

redis系列:通過demo學習string命令

redis系列:redis介紹與安裝

java多線程系列:ThreadPoolExecutor源碼分析