Java線程池的拒絕策略是什么

這篇文章主要介紹“Java線程池的拒絕策略是什么”，在日常操作中，相信很多人在Java線程池的拒絕策略是什么問題上存在疑惑，小編查閱了各式資料，整理出簡單好用的操作方法，希望對大家解答”Java線程池的拒絕策略是什么”的疑惑有所幫助！接下來，請跟著小編一起來學習吧！

池化設計思想

池話設計應該不是一個新名詞。我們常見的如java線程池、jdbc連接池、redis連接池等就是這類設計的代表實現(xiàn)。

這種設計會初始預設資源，解決的問題就是抵消每次獲取資源的消耗，如創(chuàng)建線程的開銷，獲取遠程連接的開銷等。就好比你去食堂打飯，打飯的大媽會先把飯盛好幾份放那里，你來了就直接拿著飯盒加菜即可，不用再臨時又盛飯又打菜，效率就高了。

除了初始化資源，池化設計還包括如下這些特征：池子的初始值、池子的活躍值、池子的最大值等，這些特征可以直接映射到java線程池和數(shù)據(jù)庫連接池的成員屬性中。

線程池觸發(fā)拒絕策略的時機

和數(shù)據(jù)源連接池不一樣，線程池除了初始大小和池子最大值，還多了一個阻塞隊列來緩沖。

數(shù)據(jù)源連接池一般請求的連接數(shù)超過連接池的最大值的時候就會觸發(fā)拒絕策略，策略一般是阻塞等待設置的時間或者直接拋異常。

如圖，想要了解線程池什么時候觸發(fā)拒絕粗略，需要明確上面三個參數(shù)的具體含義，是這三個參數(shù)總體協(xié)調的結果，而不是簡單的超過最大線程數(shù)就會觸發(fā)線程拒絕粗略，當提交的任務數(shù)大于corePoolSize時，會優(yōu)先放到隊列緩沖區(qū)，只有填滿了緩沖區(qū)后，才會判斷當前運行的任務是否大于maxPoolSize，小于時會新建線程處理。大于時就觸發(fā)了拒絕策略。

總結就是：當前提交任務數(shù)大于（maxPoolSize + queueCapacity）時就會觸發(fā)線程池的拒絕策略了。

JDK內置4種線程池拒絕策略

拒絕策略接口定義

在分析JDK自帶的線程池拒絕策略前，先看下JDK定義的 拒絕策略接口，如下：

public interface RejectedExecutionHandler {      void rejectedExecution(Runnable r, ThreadPoolExecutor executor);  }

接口定義很明確，當觸發(fā)拒絕策略時，線程池會調用你設置的具體的策略，將當前提交的任務以及線程池實例本身傳遞給你處理，具體作何處理，不同場景會有不同的考慮，下面看JDK為我們內置了哪些實現(xiàn)：

CallerRunsPolicy（調用者運行策略）

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

功能：當觸發(fā)拒絕策略時，只要線程池沒有關閉，就由提交任務的當前線程處理。

使用場景：一般在不允許失敗的、對性能要求不高、并發(fā)量較小的場景下使用，因為線程池一般情況下不會關閉，也就是提交的任務一定會被運行，但是由于是調用者線程自己執(zhí)行的，當多次提交任務時，就會阻塞后續(xù)任務執(zhí)行，性能和效率自然就慢了。

AbortPolicy（中止策略）

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

功能：當觸發(fā)拒絕策略時，直接拋出拒絕執(zhí)行的異常，中止策略的意思也就是打斷當前執(zhí)行流程

使用場景：這個就沒有特殊的場景了，但是一點要正確處理拋出的異常。

ThreadPoolExecutor中默認的策略就是AbortPolicy，ExecutorService接口的系列ThreadPoolExecutor因為都沒有顯示的設置拒絕策略，所以默認的都是這個。

但是請注意，ExecutorService中的線程池實例隊列都是無界的，也就是說把內存撐爆了都不會觸發(fā)拒絕策略。當自己自定義線程池實例時，使用這個策略一定要處理好觸發(fā)策略時拋的異常，因為他會打斷當前的執(zhí)行流程。

DiscardPolicy（丟棄策略）

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

功能：直接靜悄悄的丟棄這個任務，不觸發(fā)任何動作

使用場景：如果你提交的任務無關緊要，你就可以使用它 。因為它就是個空實現(xiàn)，會悄無聲息的吞噬你的的任務。所以這個策略基本上不用了

DiscardOldestPolicy（棄老策略）

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

功能：如果線程池未關閉，就彈出隊列頭部的元素，然后嘗試執(zhí)行

使用場景：這個策略還是會丟棄任務，丟棄時也是毫無聲息，但是特點是丟棄的是老的未執(zhí)行的任務，而且是待執(zhí)行優(yōu)先級較高的任務。

基于這個特性，我能想到的場景就是，發(fā)布消息，和修改消息，當消息發(fā)布出去后，還未執(zhí)行，此時更新的消息又來了，這個時候未執(zhí)行的消息的版本比現(xiàn)在提交的消息版本要低就可以被丟棄了。因為隊列中還有可能存在消息版本更低的消息會排隊執(zhí)行，所以在真正處理消息的時候一定要做好消息的版本比較。

第三方實現(xiàn)的拒絕策略

dubbo中的線程拒絕策略

public class AbortPolicyWithReport extends ThreadPoolExecutor.AbortPolicy {      protected static final Logger logger = LoggerFactory.getLogger(AbortPolicyWithReport.class);      private final String threadName;      private final URL url;      private static volatile long lastPrintTime = 0;      private static Semaphore guard = new Semaphore(1);      public AbortPolicyWithReport(String threadName, URL url) {          this.threadName = threadName;          this.url = url;      }      @Override      public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {          String msg = String.format("Thread pool is EXHAUSTED!" +                          " Thread Name: %s, Pool Size: %d (active: %d, core: %d, max: %d, largest: %d), Task: %d (completed: %d)," +                          " Executor status:(isShutdown:%s, isTerminated:%s, isTerminating:%s), in %s://%s:%d!",                  threadName, e.getPoolSize(), e.getActiveCount(), e.getCorePoolSize(), e.getMaximumPoolSize(), e.getLargestPoolSize(),                  e.getTaskCount(), e.getCompletedTaskCount(), e.isShutdown(), e.isTerminated(), e.isTerminating(),                  url.getProtocol(), url.getIp(), url.getPort());          logger.warn(msg);          dumpJStack();          throw new RejectedExecutionException(msg);      }      private void dumpJStack() {         //省略實現(xiàn)      }  }

可以看到，當dubbo的工作線程觸發(fā)了線程拒絕后，主要做了三個事情，原則就是盡量讓使用者清楚觸發(fā)線程拒絕策略的真實原因。

1）輸出了一條警告級別的日志，日志內容為線程池的詳細設置參數(shù)，以及線程池當前的狀態(tài)，還有當前拒絕任務的一些詳細信息?？梢哉f，這條日志，使用dubbo的有過生產運維經(jīng)驗的或多或少是見過的，這個日志簡直就是日志打印的典范，其他的日志打印的典范還有spring。得益于這么詳細的日志，可以很容易定位到問題所在

2）輸出當前線程堆棧詳情，這個太有用了，當你通過上面的日志信息還不能定位問題時，案發(fā)現(xiàn)場的dump線程上下文信息就是你發(fā)現(xiàn)問題的救命稻草。

3）繼續(xù)拋出拒絕執(zhí)行異常，使本次任務失敗，這個繼承了JDK默認拒絕策略的特性

Netty中的線程池拒絕策略

private static final class NewThreadRunsPolicy implements RejectedExecutionHandler {          NewThreadRunsPolicy() {              super();          }          public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {              try {                  final Thread t = new Thread(r, "Temporary task executor");                  t.start();              } catch (Throwable e) {                  throw new RejectedExecutionException(                          "Failed to start a new thread", e);              }          }      }

Netty中的實現(xiàn)很像JDK中的CallerRunsPolicy，舍不得丟棄任務。不同的是，CallerRunsPolicy是直接在調用者線程執(zhí)行的任務。而 Netty是新建了一個線程來處理的。

所以，Netty的實現(xiàn)相較于調用者執(zhí)行策略的使用面就可以擴展到支持高效率高性能的場景了。但是也要注意一點，Netty的實現(xiàn)里，在創(chuàng)建線程時未做任何的判斷約束，也就是說只要系統(tǒng)還有資源就會創(chuàng)建新的線程來處理，直到new不出新的線程了，才會拋創(chuàng)建線程失敗的異常。

activeMq中的線程池拒絕策略

new RejectedExecutionHandler() {                  @Override                  public void rejectedExecution(final Runnable r, final ThreadPoolExecutor executor) {                      try {                          executor.getQueue().offer(r, 60, TimeUnit.SECONDS);                      } catch (InterruptedException e) {                          throw new RejectedExecutionException("Interrupted waiting for BrokerService.worker");                      }                      throw new RejectedExecutionException("Timed Out while attempting to enqueue Task.");                  }              });

activeMq中的策略屬于最大努力執(zhí)行任務型，當觸發(fā)拒絕策略時，在嘗試一分鐘的時間重新將任務塞進任務隊列，當一分鐘超時還沒成功時，就拋出異常

pinpoint中的線程池拒絕策略

public class RejectedExecutionHandlerChain implements RejectedExecutionHandler {      private final RejectedExecutionHandler[] handlerChain;      public static RejectedExecutionHandler build(List<RejectedExecutionHandler> chain) {          Objects.requireNonNull(chain, "handlerChain must not be null");          RejectedExecutionHandler[] handlerChain = chain.toArray(new RejectedExecutionHandler[0]);          return new RejectedExecutionHandlerChain(handlerChain);      }      private RejectedExecutionHandlerChain(RejectedExecutionHandler[] handlerChain) {          this.handlerChain = Objects.requireNonNull(handlerChain, "handlerChain must not be null");      }      @Override      public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {          for (RejectedExecutionHandler rejectedExecutionHandler : handlerChain) {              rejectedExecutionHandler.rejectedExecution(r, executor);          }      }  }

pinpoint的拒絕策略實現(xiàn)很有特點，和其他的實現(xiàn)都不同。他定義了一個拒絕策略鏈，包裝了一個拒絕策略列表，當觸發(fā)拒絕策略時，會將策略鏈中的rejectedExecution依次執(zhí)行一遍。

到此，關于“Java線程池的拒絕策略是什么”的學習就結束了，希望能夠解決大家的疑惑。理論與實踐的搭配能更好的幫助大家學習，快去試試吧！若想繼續(xù)學習更多相關知識，請繼續(xù)關注億速云網(wǎng)站，小編會繼續(xù)努力為大家?guī)砀鄬嵱玫奈恼拢?/p>