使用ScheduledThreadPoolExecutor踩過最痛的坑

更新時間：2023年08月10日 16:57:39 作者：肥肥技術(shù)宅

這篇文章主要介紹了使用ScheduledThreadPoolExecutor踩過最痛的坑及解決方案,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教

概述

最近項目上反饋某個重要的定時任務(wù)突然不執(zhí)行了，很頭疼，開發(fā)環(huán)境和測試環(huán)境都沒有出現(xiàn)過這個問題。

定時任務(wù)采用的是 ScheduledThreadPoolExecutor ，后來一看代碼發(fā)現(xiàn)踩了一個大坑....

還原"大坑"

這個坑就是如果 ScheduledThreadPoolExecutor 中執(zhí)行的任務(wù)出錯拋出異常后，不僅不會打印異常堆棧信息，同時還會取消后面的調(diào)度

直接看例子

@Test
public void testException() throws InterruptedException {
    // 創(chuàng)建1個線程的調(diào)度任務(wù)線程池
    ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
    // 創(chuàng)建一個任務(wù)
    Runnable runnable = new Runnable() {
        volatile int num = 0;
        @Override
        public void run() {
            num ++;
            // 模擬執(zhí)行報錯
            if(num > 5) {
                throw new RuntimeException("執(zhí)行錯誤");
            }
            log.info("exec num: [{}].....", num);
        }
    };
    // 每隔1秒鐘執(zhí)行一次任務(wù)
    scheduledExecutorService.scheduleAtFixedRate(runnable, 0, 1, TimeUnit.SECONDS);
    Thread.sleep(10000);
}

運(yùn)行結(jié)果：

只執(zhí)行了5次后，就不打印，不執(zhí)行了，因為報錯了
任務(wù)報錯，也沒有打印一次堆棧，更導(dǎo)致調(diào)度任務(wù)取消，后果十分嚴(yán)重。

解決方案

解決方法也非常簡單，只要通過try catch捕獲異常即可。

運(yùn)行結(jié)果：

看到不僅打印了異常堆棧，而且也會進(jìn)行周期性的調(diào)度。

更推薦的做法

更好的建議可以在自己的項目中封裝一個包裝類，要求所有的調(diào)度都提交通過我們統(tǒng)一的包裝類，如下代碼：

@Slf4j
public class RunnableWrapper implements Runnable {
    // 實際要執(zhí)行的線程任務(wù)
    private Runnable task;
    // 線程任務(wù)被創(chuàng)建出來的時間
    private long createTime;
    // 線程任務(wù)被線程池運(yùn)行的開始時間
    private long startTime;
    // 線程任務(wù)被線程池運(yùn)行的結(jié)束時間
    private long endTime;
    // 線程信息
    private String taskInfo;
    private boolean showWaitLog;
    /**
     * 執(zhí)行間隔時間多久，打印日志
     */
    private long durMs = 1000L;
    // 當(dāng)這個任務(wù)被創(chuàng)建出來的時候，就會設(shè)置他的創(chuàng)建時間
    // 但是接下來有可能這個任務(wù)提交到線程池后，會進(jìn)入線程池的隊列排隊
    public RunnableWrapper(Runnable task, String taskInfo) {
        this.task = task;
        this.taskInfo = taskInfo;
        this.createTime = System.currentTimeMillis();
    }
    public void setShowWaitLog(boolean showWaitLog) {
        this.showWaitLog = showWaitLog;
    }
    public void setDurMs(long durMs) {
        this.durMs = durMs;
    }
    // 當(dāng)任務(wù)在線程池排隊的時候，這個run方法是不會被運(yùn)行的
    // 但是當(dāng)任務(wù)結(jié)束了排隊，得到線程池運(yùn)行機(jī)會的時候，這個方法會被調(diào)用
    // 此時就可以設(shè)置線程任務(wù)的開始運(yùn)行時間
    @Override
    public void run() {
        this.startTime = System.currentTimeMillis();
        // 此處可以通過調(diào)用監(jiān)控系統(tǒng)的API，實現(xiàn)監(jiān)控指標(biāo)上報
        // 用線程任務(wù)的startTime-createTime，其實就是任務(wù)排隊時間
        // 這邊打印日志輸出，也可以輸出到監(jiān)控系統(tǒng)中
        if(showWaitLog) {
            log.info("任務(wù)信息: [{}], 任務(wù)排隊時間: [{}]ms", taskInfo, startTime - createTime);
        }
        // 接著可以調(diào)用包裝的實際任務(wù)的run方法
        try {
            task.run();
        } catch (Exception e) {
            log.error("run task error", e);
            throw e;
        }
        // 任務(wù)運(yùn)行完畢以后，會設(shè)置任務(wù)運(yùn)行結(jié)束的時間
        this.endTime = System.currentTimeMillis();
        // 此處可以通過調(diào)用監(jiān)控系統(tǒng)的API，實現(xiàn)監(jiān)控指標(biāo)上報
        // 用線程任務(wù)的endTime - startTime，其實就是任務(wù)運(yùn)行時間
        // 這邊打印任務(wù)執(zhí)行時間，也可以輸出到監(jiān)控系統(tǒng)中
        if(endTime - startTime > durMs) {
            log.info("任務(wù)信息: [{}], 任務(wù)執(zhí)行時間: [{}]ms", taskInfo, endTime - startTime);
        }
    }
}

使用：

我們還可以在包裝類里面封裝各種監(jiān)控行為，如本例打印日志執(zhí)行時間等。

原理探究

那大家有沒有想過為什么任務(wù)出錯會導(dǎo)致異常無法打印，甚至調(diào)度都取消了呢？讓我們從源碼出發(fā)，一探究竟。

下面是調(diào)度任務(wù)的入口方法

// ScheduledThreadPoolExecutor#scheduleAtFixedRate
public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
                                              long initialDelay,
                                              long period,
                                              TimeUnit unit) {
    if (command == null || unit == null)
        throw new NullPointerException();
    if (period <= 0)
        throw new IllegalArgumentException();
    // 將執(zhí)行任務(wù)和參數(shù)包裝成ScheduledFutureTask對象
    ScheduledFutureTask<Void> sft =
        new ScheduledFutureTask<Void>(command,
                                      null,
                                      triggerTime(initialDelay, unit),
                                      unit.toNanos(period));
    RunnableScheduledFuture<Void> t = decorateTask(command, sft);
    sft.outerTask = t;
    // 延遲執(zhí)行
    delayedExecute(t);
    return t;
}
復(fù)制代碼

這個方法主要做了兩個事情

將執(zhí)行任務(wù)和參數(shù)包裝成ScheduledFutureTask對象
調(diào)用delayedExecute方法延遲執(zhí)行任務(wù)

延遲或周期性任務(wù)的主要執(zhí)行方法，主要是將任務(wù)丟到隊列中，后續(xù)由工作線程獲取執(zhí)行。

// ScheduledThreadPoolExecutor#delayedExecute
private void delayedExecute(RunnableScheduledFuture<?> task) {
        if (isShutdown())
            reject(task);
        else {
            // 將任務(wù)丟到阻塞隊列中
            super.getQueue().add(task);
            if (isShutdown() &&
                !canRunInCurrentRunState(task.isPeriodic()) &&
                remove(task))
                task.cancel(false);
            else
                // 開啟工作線程，去執(zhí)行任務(wù)，或者從隊列中獲取任務(wù)執(zhí)行
                ensurePrestart();
        }
    }

現(xiàn)在任務(wù)已經(jīng)在隊列中了，我們看下任務(wù)執(zhí)行的內(nèi)容是什么，還記得前面的包裝對象 ScheduledFutureTask 類，它的實現(xiàn)類是 ScheduledFutureTask ，繼承了Runnable類。

// ScheduledFutureTask#run方法
public void run() {
    // 是不是周期性任務(wù)
    boolean periodic = isPeriodic();
    if (!canRunInCurrentRunState(periodic))
        cancel(false);
    // 不是周期性任務(wù)的話， 直接調(diào)用一次下面的run    
    else if (!periodic)
        ScheduledFutureTask.super.run();
    // 如果是周期性任務(wù)，則調(diào)用runAndReset方法，如果返回true，繼續(xù)執(zhí)行
    else if (ScheduledFutureTask.super.runAndReset()) {
        // 設(shè)置下次調(diào)度時間
        setNextRunTime();
        // 重新執(zhí)行調(diào)度任務(wù)
        reExecutePeriodic(outerTask);
    }
}

這里的關(guān)鍵就是看 ScheduledFutureTask.super.runAndReset() 方法是否返回true，如果是true的話繼續(xù)調(diào)度。

runAndReset方法也很簡單，關(guān)鍵就是看報異常如何處理。

// FutureTask#runAndReset
protected boolean runAndReset() {
    if (state != NEW ||
        !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                     null, Thread.currentThread()))
        return false;
    // 是否繼續(xù)下次調(diào)度，默認(rèn)false
    boolean ran = false;
    int s = state;
    try {
        Callable<V> c = callable;
        if (c != null && s == NEW) {
            try {
                // 執(zhí)行任務(wù)
                c.call(); 
                // 執(zhí)行成功的話，設(shè)置為true
                ran = true;
                // 異常處理，關(guān)鍵點
            } catch (Throwable ex) {
                // 不會修改ran的值，最終是false，同時也不打印異常堆棧
                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);
    }
    // 返回結(jié)果
    return ran && s == NEW;
}