線程池 ThreadPoolExecutor 為了提供擴(kuò)展，提供了兩個方法 beforeExecute 和 afterExecute，每個任務(wù)執(zhí)行前后都會調(diào)用這兩個方法，相當(dāng)于對線程任務(wù)的執(zhí)行做了一個切面。
?監(jiān)控線程池：執(zhí)行超時、等待超時；執(zhí)行超時數(shù)量、等待超時數(shù)量；
擴(kuò)展線程池 ThreadPoolExecutor 的兩個方法 beforeExecute 和 afterExecute
自定義Runnable 記錄關(guān)鍵節(jié)點時間
關(guān)鍵時間節(jié)點參數(shù)：
- 任務(wù)創(chuàng)建(提交)時間：submitTime
- 任務(wù)開始執(zhí)行時間：startExeTime
- 任務(wù)結(jié)束執(zhí)行時間：endExeTime
- 任務(wù)在隊列等待時間：任務(wù)開始執(zhí)行時間 - 任務(wù)創(chuàng)建(提交)時間
- 任務(wù)執(zhí)行總時間：任務(wù)結(jié)束執(zhí)行時間 - 任務(wù)開始執(zhí)行時間
源碼分析
線程池 ThreadPoolExecutor 為了提供擴(kuò)展，提供了兩個方法 beforeExecute 和 afterExecute，每個任務(wù)執(zhí)行前后都會調(diào)用這兩個方法，相當(dāng)于對線程任務(wù)的執(zhí)行做了一個切面。
public class ThreadPoolExecutor extends AbstractExecutorService {
/**
* @param t 執(zhí)行任務(wù)的線程
* @param
protected void beforeExecute(Thread t, Runnable r){ }
/**
* @param r 將要被執(zhí)行的任務(wù)
* @param
protected void afterExecute(Runnable r, Throwable t){ }
}源碼執(zhí)行邏輯：
線程池擴(kuò)展代碼：
public class ThreadPoolExpandTest {
// 定義線程池
public static ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(
2,
4,
60,
TimeUnit.SECONDS,
new ArrayBlockingQueue<>(5),
new ThreadPoolExecutor.DiscardOldestPolicy()
){
@Override
/**
* @param t 執(zhí)行任務(wù)的線程
* @param
protected void beforeExecute(Thread t, Runnable r){
System.out.println("beforeExecute將要被執(zhí)行");
}
/**
* @param r 將要被執(zhí)行的任務(wù)
* @param
@Override
protected void afterExecute(Runnable r, Throwable t){
System.out.println("afterExecute已經(jīng)執(zhí)行完畢");
}
};
public static void main(String[] args){
poolExecutor.execute(()->{
System.out.println("任務(wù)執(zhí)行");
});
}
}運行結(jié)果：
beforeExecute執(zhí)行
任務(wù)執(zhí)行
afterExecute執(zhí)行
總結(jié)：從測試代碼可以看出，通過擴(kuò)展線程池參數(shù)可以進(jìn)行任務(wù)執(zhí)行的監(jiān)控。
自定義Runnable
通過自定義Runnable，記錄任務(wù)執(zhí)行的一些時間：
- 任務(wù)創(chuàng)建(提交)時間
- 任務(wù)開始執(zhí)行時間
public class DynamicRunnable implements Runnable{
/**
* runnable
*/
private final Runnable runnable;
/**
* 任務(wù)創(chuàng)建(提交)時間
*/
private final Long submitTime;
/**
* 任務(wù)開始執(zhí)行時間
*/
private Long startExeTime;
public DynamicRunnable(Runnable runnable){
this.runnable = runnable;
submitTime = System.currentTimeMillis();
}
@Override
public void run(){
runnable.run();
}
public Long getSubmitTime(){
return submitTime;
}
public void setStartExeTime(Long startExeTime){
this.startExeTime = startExeTime;
}
public Long getStartExeTime(){
return startExeTime;
}
}繼承線程池+自定義Runnable
核心參數(shù)：
/**
* 執(zhí)行超時，單位（毫秒）
*/
private long runTimeout;
/**
* 等待超時，單位（毫秒）
*/
private long queueTimeout;
/**
* 執(zhí)行超時數(shù)量
*/
private final AtomicInteger runTimeoutCount = new AtomicInteger();
/**
* 等待超時數(shù)量
*/
private final AtomicInteger queueTimeoutCount = new AtomicInteger();
重寫ThreadPoolExecutor方法：
@Override
public void execute(Runnable command){
if (runTimeout > 0 || queueTimeout > 0) {
// 記錄任務(wù)提交時間
command = new DynamicRunnable(command);
}
super.execute(command);
}
@Override
protected void beforeExecute(Thread t, Runnable r){
if (!(r instanceof DynamicRunnable)) {
super.beforeExecute(t, r);
return;
}
DynamicRunnable runnable = (DynamicRunnable) r;
long currTime = System.currentTimeMillis();
if (runTimeout > 0) {
// 記錄任務(wù)開始執(zhí)行時間
runnable.setStartExeTime(currTime);
}
if (queueTimeout > 0) {
// 任務(wù)開始執(zhí)行時間 - 任務(wù)創(chuàng)建(提交)時間
long waitTime = currTime - runnable.getSubmitTime();
if (waitTime > queueTimeout) {
log.error("{} execute queue timeout waitTime: {}ms", this.getThreadPoolName(),waitTime);
}
}
super.beforeExecute(t, r);
}
@Override
protected void afterExecute(Runnable r, Throwable t){
if (runTimeout > 0) {
DynamicRunnable runnable = (DynamicRunnable) r;
// 任務(wù)執(zhí)行總時間：任務(wù)結(jié)束執(zhí)行時間 - 任務(wù)開始執(zhí)行時間
long runTime = System.currentTimeMillis() - runnable.getStartExeTime();
if (runTime > runTimeout) {
runTimeoutCount.incrementAndGet();
log.error("{} execute, run timeout runTime: {}ms", this.getThreadPoolName(), runTime);
}
}
super.afterExecute(r, t);
}
