ScheduledThreadPoolExecutor巨坑解決

更新時間：2023年02月22日 11:15:10 作者：Code皮皮蝦

這篇文章主要為大家介紹了使用ScheduledThreadPoolExecutor遇到的巨坑解決示例，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪

概述

最近在做一些優(yōu)化的時候用到了ScheduledThreadPoolExecutor。

雖然知道這個玩意，但是也很久沒用，本著再了解了解的心態(tài)，到網(wǎng)上搜索了一下，結(jié)果就發(fā)現(xiàn)網(wǎng)上有些博客在說ScheduledThreadPoolExecutor有巨坑?。?！

瞬間,我的興趣就被激起來了，馬上進(jìn)去學(xué)習(xí)了一波～

不看不知道，看完后馬上把我的代碼坑給填上了～

下面就當(dāng)記錄一下吧，順便也帶大家了解了解，大家看完后也趕緊看看自己公司的項目代碼有沒有這種漏洞，有的話趕緊給填上，升級加薪指日可待?。?！

坑是啥？

先看下面案例代碼

public class ScheduledThreadPoolExecutorTest {
  public static ScheduledThreadPoolExecutor scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(2);
  public static AtomicInteger atomicInteger = new AtomicInteger(1);
  public static void main(String[] args) {
    scheduledThreadPoolExecutor.scheduleAtFixedRate(() -> {
      // 模擬業(yè)務(wù)邏輯
      int num = atomicInteger.getAndIncrement();
      // 模擬出現(xiàn)異常
      if (num > 3) {
        throw new RuntimeException("定時任務(wù)執(zhí)行異常");
      }
      System.out.println("別坑我！");
    }, 0, 1, TimeUnit.SECONDS);
    try {
      TimeUnit.SECONDS.sleep(5);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    scheduledThreadPoolExecutor.shutdown();
  }
}

案例代碼邏輯很簡單，主線程等待5秒后關(guān)閉線程池，定時任務(wù)執(zhí)行三次后模擬拋出RuntimeException

但是我們看看執(zhí)行結(jié)果,只執(zhí)行了三次！

因為某種情況下，定時任務(wù)在執(zhí)行第四次時出現(xiàn)異常，從而導(dǎo)致任務(wù)調(diào)度被取消，不會繼續(xù)執(zhí)行

而且，異常信息也沒有對外拋出！

那么咋解決嘞？try-catch就行了唄～

可以看到執(zhí)行結(jié)果，雖然執(zhí)行異常，但是任務(wù)卻還是一直在調(diào)度～

代碼里使用工具類對Runnable任務(wù)包了一層，就是加了try-catch

public class RunnableDecoratorUtil {
   public static Runnable runnableDecorator(Runnable runnable) {
      return () -> {
         try {
            runnable.run();
         } catch (Exception e) {
            e.printStackTrace();
         }
      };
   }
}

okok，總結(jié)一下，坑就是: 任務(wù)如果拋出異常就不會繼續(xù)調(diào)度執(zhí)行了，趕緊去try-catch吧?。?！

大家趕緊去看看自己代碼有沒有這個坑吧，本文到此結(jié)束！

開個玩笑～光知道有坑哪能不知道為啥坑，接下來就帶大家了解一下坑到底是啥！

怎么坑的？

直接進(jìn)入scheduleAtFixedRate源碼查看

public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
                                              long initialDelay,
                                              long period,
                                              TimeUnit unit) {
    // 參數(shù)校驗
    if (command == null || unit == null)
        throw new NullPointerException();
    if (period <= 0L)
        throw new IllegalArgumentException();
    // 將任務(wù)、執(zhí)行時間、周期等封裝到ScheduledFutureTask內(nèi)
    ScheduledFutureTask<Void> sft =
        new ScheduledFutureTask<Void>(command,
                                      null,
                                      triggerTime(initialDelay, unit),
                                      unit.toNanos(period),
                                      sequencer.getAndIncrement());
    RunnableScheduledFuture<Void> t = decorateTask(command, sft);
    sft.outerTask = t;
    // 延時執(zhí)行
    delayedExecute(t);
    return t;
}

因為我們提交的任務(wù)被封裝在ScheduledFutureTask，所以我們直接來看ScheduledFutureTask的run方法

public void run() {
  // 校驗當(dāng)前狀態(tài)是否還能執(zhí)行任務(wù)，不能執(zhí)行直接cancel取消
  if (!canRunInCurrentRunState(this))
    cancel(false);
  else if (!isPeriodic())
    // 如果不是周期性的，直接調(diào)用父類run方法執(zhí)行一次即可
    super.run();
  else if (super.runAndReset()) { // 周期性任務(wù)，調(diào)用runAndReset運行并重置
    // 設(shè)置下一次的執(zhí)行時間
    setNextRunTime();
    // 將任務(wù)重新加入隊列，進(jìn)行調(diào)度
    reExecutePeriodic(outerTask);
  }
}
public boolean isPeriodic() {
  return period != 0;
}

我們是周期性任務(wù)，所以直接看runAndReset源碼

protected boolean runAndReset() {
    // 檢查任務(wù)狀態(tài)，cas機(jī)制防止并發(fā)執(zhí)行任務(wù)
    if (state != NEW ||
        !RUNNER.compareAndSet(this, null, Thread.currentThread()))
        return false;
    // 默認(rèn)不周期執(zhí)行任務(wù)
    boolean ran = false;
    // state為NEW狀態(tài)
    int s = state;
    try {
        Callable<V> c = callable;
        if (c != null && s == NEW) {
            try {
                // 執(zhí)行任務(wù)
                c.call();
                // 正常執(zhí)行成功，設(shè)置為true代表周期執(zhí)行
                ran = true;
            } catch (Throwable ex) {
                // 但是，如果執(zhí)行異常！則不會將ran = true，所以最終返回false
                setException(ex);
            }
        }
    } finally {
        runner = null;
        // 設(shè)置為NEW狀態(tài)
        s = state;
        if (s >= INTERRUPTING)
            handlePossibleCancellationInterrupt(s);
    }
    // 正常執(zhí)行完之后，結(jié)果為true，能夠周期執(zhí)行
    // 但如果執(zhí)行異常，ran為false，返回結(jié)果為false
    return ran && s == NEW;
}

通過上面源碼，我們可以很清楚的了解到，就是因為任務(wù)執(zhí)行異常，且沒有被try-catch，所以導(dǎo)致任務(wù)沒有被再次加入到隊列中進(jìn)行調(diào)度。

并且通過文章開頭，我們還能看到任務(wù)執(zhí)行異常，但是卻沒有拋出異常信息

那是因為異常被封裝了，只有調(diào)用get方法時，才會拋出異常

/** The result to return or exception to throw from get() */
private Object outcome;
private volatile int state;
private static final int NEW          = 0;
private static final int COMPLETING   = 1;
private static final int NORMAL       = 2;
private static final int EXCEPTIONAL  = 3;
private static final int CANCELLED    = 4;
protected void setException(Throwable t) {
    if (STATE.compareAndSet(this, NEW, COMPLETING)) {
        // 將異常信息賦值給outcome
       // outcome既可以為任務(wù)執(zhí)行結(jié)果也可以為異常信息
        outcome = t;
        // 將state設(shè)置為異常狀態(tài)，state=3
        STATE.setRelease(this, EXCEPTIONAL); // final state
        finishCompletion();
    }
}
// 調(diào)用get方法阻塞獲取結(jié)果
public V get() throws InterruptedException, ExecutionException {
  int s = state;
  if (s <= COMPLETING)
    s = awaitDone(false, 0L);
  return report(s);
}
private V report(int s) throws ExecutionException {
  Object x = outcome;
  // 此時s = EXCEPTIONAL = 3
  if (s == NORMAL)
    return (V)x;
  if (s >= CANCELLED)
    throw new CancellationException();
  // 所以會走到這里，對外拋出了任務(wù)執(zhí)行的異常
  throw new ExecutionException((Throwable)x);
}