關于ScheduledThreadPoolExecutor不執(zhí)行的原因分析
ScheduledThreadPoolExecutor不執(zhí)行原因分析
最近在調(diào)試一個監(jiān)控應用指標的時候發(fā)現(xiàn)定時器在服務啟動執(zhí)行一次之后就不執(zhí)行了，這里用的定時器是Java的調(diào)度線程池 ScheduledThreadPoolExecutor ，后來經(jīng)過排查發(fā)現(xiàn) ScheduledThreadPoolExecutor 線程池處理任務如果拋出異常，會導致線程池不調(diào)度；下面就通過一個例子簡單分析下為什么異常會導致 ScheduledThreadPoolExecutor 不執(zhí)行。
ScheduledThreadPoolExecutor不調(diào)度分析
示例程序
在示例程序可以看到當計數(shù)器中的計數(shù)達到5的時候就會主動拋出一個異常，拋出異常后 ScheduledThreadPoolExecutor 就不調(diào)度了。
public class ScheduledTask {
private static final AtomicInteger count = new AtomicInteger(0);
private static final ScheduledThreadPoolExecutor SCHEDULED_TASK = new ScheduledThreadPoolExecutor(
1, new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
Thread t = new Thread(Thread.currentThread().getThreadGroup(), r, "sc-task");
t.setDaemon(true);
return t;
}
});
public static void main(String[] args) throws InterruptedException {
CountDownLatch latch = new CountDownLatch(1);
SCHEDULED_TASK.scheduleWithFixedDelay(() -> {
System.out.println(111);
if (count.get() == 5) {
throw new IllegalArgumentException("my exception");
}
count.incrementAndGet();
}, 0, 5, TimeUnit.SECONDS);
latch.await();
}
}源碼分析
- ScheduledThreadPoolExecutor#run
run方法內(nèi)部首先判斷任務是不是周期性的任務，如果不是周期性任務通過 ScheduledFutureTask.super.run(); 執(zhí)行任務；如果狀態(tài)是運行中或shutdown，取消任務執(zhí)行；如果是周期性的任務，通過 ScheduledFutureTask.super.runAndReset() 執(zhí)行任務并且重新設置狀態(tài)，成功了就會執(zhí)行 setNextRunTime 設置下次調(diào)度的時間，問題就是出現(xiàn)在 ScheduledFutureTask.super.runAndReset() ，這里執(zhí)行任務出現(xiàn)了異常，導致結(jié)果為false，就不進行下次調(diào)度時間設置等
public void run() {
boolean periodic = isPeriodic();
if (!canRunInCurrentRunState(periodic))
cancel(false);
else if (!periodic)
ScheduledFutureTask.super.run();
else if (ScheduledFutureTask.super.runAndReset()) {
setNextRunTime();
reExecutePeriodic(outerTask);
}
}- *FutureTask#runAndReset
在線程任務執(zhí)行過程中拋出異常，然后 catch 到了異常，最終導致這個方法返回false，然后 ScheduledThreadPoolExecutor#run 就不設置下次執(zhí)行時間了，代碼 c.call(); 拋出異常，跳過 ran = true; 代碼，最終 runAndReset 返回false。
protected boolean runAndReset() {
if (state != NEW ||
!UNSAFE.compareAndSwapObject(this, runnerOffset,
null, Thread.currentThread()))
return false;
boolean ran = false;
int s = state;
try {
Callable<V> c = callable;
if (c != null && s == NEW) {
try {
c.call(); // don't set result
ran = true;
} catch (Throwable ex) {
setException(ex);
}
}
} finally {
// runner must be non-null until state is settled to
// prevent concurrent calls to run()
runner = null;
// state must be re-read after nulling runner to prevent
// leaked interrupts
s = state;
if (s >= INTERRUPTING)
handlePossibleCancellationInterrupt(s);
}
return ran && s == NEW;
}注意：
Java 的 ScheduledThreadPoolExecutor 定時任務線程池所調(diào)度的任務中如果拋出了異常，并且異常沒有捕獲直接拋到框架中，會導致 ScheduledThreadPoolExecutor 定時任務不調(diào)度了，具體是因為當異常拋到 ScheduledThreadPoolExecutor 框架中時不進行下次調(diào)度時間的設置，從而導致 ScheduledThreadPoolExecutor 定時任務不調(diào)度。
ScheduledThreadPoolExecutor線程池例子
ScheduledThreadPoolExecutor 使用
ScheduledThreadPoolExecutor 繼承自 ThreadPoolExecutor，主要用來給定時間運行任務，或者定期執(zhí)行任務。
在 ScheduledThreadPoolExecutor 的實現(xiàn)中，使用了 FutureTask 運行任務以及使用無界隊列 DelayedWorkQueue 來保存任務。
1. 使用示例
- 提交任務
ScheduledThreadPoolExecutor 實現(xiàn)了 ScheduledExecutorService 接口，其中，接口中有四個需要實現(xiàn)的方法，其中 schedule() 的兩個方法需要設置任務以及任務啟動的延遲時間，scheduleAtFixedRate() 可以設置任務定時重復執(zhí)行，scheduleWithFixedDelay() 則是設置兩個任務之間的執(zhí)行延遲時間。
ScheduledThreadPoolExecutor poolExecutor = new ScheduledThreadPoolExecutor(2); ?
poolExecutor.schedule(() -> { ?
?? ?// 提交的任務
}, 5, TimeUnit.HOURS); ?
poolExecutor.scheduleAtFixedRate(() -> { ?
?? ?// 提交的任務
}, 0, 5, TimeUnit.HOURS); ?
poolExecutor.scheduleWithFixedDelay(() -> { ?
?? ?// 提交的任務
}, 0, 5, TimeUnit.HOURS);- 簡單例子
下面的例子中每 500 毫秒打印一次字符串，executor 會有 5 秒的時間來等待任務執(zhí)行結(jié)束，也就是一共可以打印 10 次字符串。
ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(10); ?
executor.scheduleWithFixedDelay(() -> { ?
? ? System.out.println("測試"); ?
}, 0, 500, TimeUnit.MILLISECONDS); ?
try { ?
? ? executor.awaitTermination(5, TimeUnit.SECONDS); ?
? ? executor.shutdown(); ?
} catch (InterruptedException e) { ?
? ? e.printStackTrace(); ?
}ScheduledThreadPoolExecutor 原理
1. DelayedWorkQueue
ScheduledThreadPoolExecutor 的構(gòu)造方法對 DelayedWorkQueue 進行了初始化，并且最大線程數(shù)量設置成了 Integer.MAX_VALUE。
public ScheduledThreadPoolExecutor(int corePoolSize) { ?
? ? super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS, ?
? ? ? ? ? new DelayedWorkQueue()); ?
}其中，DelayedWorkQueue 中的隊列是 RunnableScheduledFuture 類型以及容量為 16 的數(shù)組。
private static final int INITIAL_CAPACITY = 16; ? private RunnableScheduledFuture<?>[] queue = ? ? ? new RunnableScheduledFuture<?>[INITIAL_CAPACITY]; ? private final ReentrantLock lock = new ReentrantLock();
隊列添加任務是以 DelayedWorkQueue 以堆作為數(shù)據(jù)結(jié)構(gòu)存儲任務，在添加元素的時候，會使用基于 Siftup 版本進行元素添加，并且會根據(jù)任務的執(zhí)行時間的大小來排序。
public boolean offer(Runnable x) { ?
? ? if (x == null) ?
? ? ? ? throw new NullPointerException();
? ? RunnableScheduledFuture<?> e = (RunnableScheduledFuture<?>)x; ?
? ? final ReentrantLock lock = this.lock; ?
? ? lock.lock();
? ? try {
?? ? ? ?// 當隊列的容量不夠，會擴充 50%
? ? ? ? int i = size;
? ? ? ? if (i >= queue.length)
? ? ? ? ? ? grow();
? ? ? ? size = i + 1;
? ? ? ? if (i == 0) {
? ? ? ? ? ? queue[0] = e;
? ? ? ? ? ? setIndex(e, 0);
? ? ? ? } else { ?
? ? ? ? ? ? siftUp(i, e); ?
? ? ? ? } ?
? ? ? ? if (queue[0] == e) { ?
? ? ? ? ? ? leader = null; ?
? ? ? ? ? ? available.signal(); ?
? ? ? ? } ?
? ? } finally { ?
? ? ? ? lock.unlock(); ?
? ? } ?
? ? return true; ?
}2. delayedExecute()
ScheduledExecutorService 接口的四個實現(xiàn)方法中都涉及到了 delayedExecute()，方法主要用來判斷線程池的狀態(tài)以及對線程進行初始化。
private void delayedExecute(RunnableScheduledFuture<?> task) {
?? ?// 如果線程池關閉了，需要執(zhí)行飽和策略
? ? if (isShutdown()) ?
? ? ? ? reject(task); ?
? ? else {
?? ? ? ?// 添加到隊列中
? ? ? ? super.getQueue().add(task); ?
? ? ? ? if (isShutdown() && ?
? ? ? ? ? ? !canRunInCurrentRunState(task.isPeriodic()) && ?
? ? ? ? ? ? remove(task)) ?
? ? ? ? ? ? task.cancel(false); ?
? ? ? ? else
?? ? ? ? ? ?// 判斷等待隊列中是否已經(jīng)滿了，會使用到 ThreadPoolExecutor
?? ? ? ? ? ?ensurePrestart(); ?
? ? } ?
}
void ensurePrestart() { ?
? ? int wc = workerCountOf(ctl.get()); ?
? ? if (wc < corePoolSize) ?
? ? ? ? addWorker(null, true); ?
? ? else if (wc == 0) ?
? ? ? ? addWorker(null, false); ?
}總結(jié)
以上為個人經(jīng)驗，希望能給大家一個參考，也希望大家多多支持腳本之家。
相關文章
這篇文章主要介紹了IDEA創(chuàng)建parent項目(聚合項目),文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2019-08-08
這篇文章主要介紹了springboot中的RestTemplate使用詳解,RestTemplate繼承自InterceptingHttpAccessor并且實現(xiàn)了RestOperations接口,其中RestOperations接口定義了基本的RESTful操作,這些操作在RestTemplate中都得到了實現(xiàn),需要的朋友可以參考下2023-09-09
Spring boot如何配置請求的入?yún)⒑统鰠son數(shù)據(jù)格式
這篇文章主要介紹了spring boot如何配置請求的入?yún)⒑统鰠son數(shù)據(jù)格式,文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下2019-11-11
Java中l(wèi)og4j注解詳解與實戰(zhàn)指南(附具體示例)
日志是軟件開發(fā)中不可或缺的一部分,它能夠幫助開發(fā)者追蹤程序的運行狀態(tài),診斷問題,以及監(jiān)控系統(tǒng)性能,這篇文章主要介紹了Java中l(wèi)og4j注解詳解與實戰(zhàn)的相關資料,需要的朋友可以參考下2025-07-07
這篇文章主要給大家總結(jié)介紹了關于Java開發(fā)中最讓人頭疼的十個bug,同樣的bug信息,可能背后有千萬種原因,而我,永遠都不知道到底是哪一個,努力通過代碼積累盡可能多的bug,并將它們進行分類,可以幫你debug節(jié)省了時間,需要的朋友可以參考下2021-10-10
這篇文章主要介紹了SpringBoot全局異常處理與定制404頁面的相關資料,本文通過實例代碼圖文相結(jié)合給大家介紹的非常詳細，需要的朋友可以參考下2007-09-09
一文詳解Spring中ResponseEntity包裝器的使用
在?Spring?中,ResponseEntity?是?HTTP?響應的包裝器,這篇文章主要為大家詳細介紹了ResponseEntity包裝器的使用,感興趣的可以了解一下2025-02-02
這篇文章主要介紹了如何解決IDEA.properties文件中文亂碼問題,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2023-12-12
SpringBoot根據(jù)目錄結(jié)構(gòu)自動生成路由前綴的實現(xiàn)代碼
本文介紹如何根據(jù)目錄結(jié)構(gòu)給RequestMapping添加路由前綴，具體實現(xiàn)方法，本文通過示例代碼給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值，需要的朋友參考下吧2021-08-08
二叉樹是一種很有用的非線性結(jié)構(gòu),日常的開發(fā)中常會用到，關于二叉樹的概念以及搜索二叉樹本文給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值，需要的朋友可以參考下2021-09-09