java線程池ThreadPoolExecutor八種拒絕策略淺析

時間 2020-02-01

標籤 java 線程 threadpoolexecutor 拒絕策略淺析欄目 Java 简体版

原文原文鏈接

前言

談到java的線程池最熟悉的莫過於ExecutorService接口了，jdk1.5新增的java.util.concurrent包下的這個api，大大的簡化了多線程代碼的開發。而不論你用FixedThreadPool仍是CachedThreadPool其背後實現都是ThreadPoolExecutor。ThreadPoolExecutor是一個典型的緩存池化設計的產物，由於池子有大小，當池子體積不夠承載時，就涉及到拒絕策略。JDK中已經預設了4種線程池拒絕策略，下面結合場景詳細聊聊這些策略的使用場景，以及咱們還能擴展哪些拒絕策略。java

池化設計思想

池話設計應該不是一個新名詞。咱們常見的如java線程池、jdbc鏈接池、redis鏈接池等就是這類設計的表明實現。這種設計會初始預設資源，解決的問題就是抵消每次獲取資源的消耗，如建立線程的開銷，獲取遠程鏈接的開銷等。就比如你去食堂打飯，打飯的大媽會先把飯盛好幾份放那裏，你來了就直接拿着飯盒加菜便可，不用再臨時又盛飯又打菜，效率就高了。除了初始化資源，池化設計還包括以下這些特徵：池子的初始值、池子的活躍值、池子的最大值等，這些特徵能夠直接映射到java線程池和數據庫鏈接池的成員屬性中。redis

線程池觸發拒絕策略的時機

和數據源鏈接池不同，線程池除了初始大小和池子最大值，還多了一個阻塞隊列來緩衝。數據源鏈接池通常請求的鏈接數超過鏈接池的最大值的時候就會觸發拒絕策略，策略通常是阻塞等待設置的時間或者直接拋異常。而線程池的觸發時機以下圖：spring

如圖，想要了解線程池何時觸發拒絕粗略，須要明確上面三個參數的具體含義，是這三個參數整體協調的結果，而不是簡單的超過最大線程數就會觸發線程拒絕粗略，當提交的任務數大於corePoolSize時，會優先放到隊列緩衝區，只有填滿了緩衝區後，纔會判斷當前運行的任務是否大於maxPoolSize，小於時會新建線程處理。大於時就觸發了拒絕策略，總結就是：當前提交任務數大於（maxPoolSize + queueCapacity）時就會觸發線程池的拒絕策略了。數據庫

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

拒絕策略接口定義

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

public interface RejectedExecutionHandler {
    void rejectedExecution(Runnable r, ThreadPoolExecutor executor);
}複製代碼

接口定義很明確，當觸發拒絕策略時，線程池會調用你設置的具體的策略，將當前提交的任務以及線程池實例自己傳遞給你處理，具體做何處理，不一樣場景會有不一樣的考慮，下面看JDK爲咱們內置了哪些實現：緩存

CallerRunsPolicy（調用者運行策略）

public static class CallerRunsPolicy implements RejectedExecutionHandler {

        public CallerRunsPolicy() { }

        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            if (!e.isShutdown()) {
                r.run();
            }
        }
    }複製代碼

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

使用場景：通常在不容許失敗的、對性能要求不高、併發量較小的場景下使用，由於線程池通常狀況下不會關閉，也就是提交的任務必定會被運行，可是因爲是調用者線程本身執行的，當屢次提交任務時，就會阻塞後續任務執行，性能和效率天然就慢了。微信

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());
        }
    }複製代碼

功能：當觸發拒絕策略時，直接拋出拒絕執行的異常，停止策略的意思也就是打斷當前執行流程多線程

使用場景：這個就沒有特殊的場景了，可是一點要正確處理拋出的異常。ThreadPoolExecutor中默認的策略就是AbortPolicy，ExecutorService接口的系列ThreadPoolExecutor由於都沒有顯示的設置拒絕策略，因此默認的都是這個。可是請注意，ExecutorService中的線程池實例隊列都是無界的，也就是說把內存撐爆了都不會觸發拒絕策略。當本身自定義線程池實例時，使用這個策略必定要處理好觸發策略時拋的異常，由於他會打斷當前的執行流程。併發

DiscardPolicy（丟棄策略）

public static class DiscardPolicy implements RejectedExecutionHandler {

        public DiscardPolicy() { }

        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        }
    }複製代碼

功能：直接靜悄悄的丟棄這個任務，不觸發任何動做

使用場景：若是你提交的任務可有可無，你就可使用它。由於它就是個空實現，會悄無聲息的吞噬你的的任務。因此這個策略基本上不用了

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);
            }
        }
    }複製代碼

功能：若是線程池未關閉，就彈出隊列頭部的元素，而後嘗試執行

使用場景：這個策略仍是會丟棄任務，丟棄時也是毫無聲息，可是特色是丟棄的是老的未執行的任務，並且是待執行優先級較高的任務。基於這個特性，我能想到的場景就是，發佈消息，和修改消息，當消息發佈出去後，還未執行，此時更新的消息又來了，這個時候未執行的消息的版本比如今提交的消息版本要低就能夠被丟棄了。由於隊列中還有可能存在消息版本更低的消息會排隊執行，因此在真正處理消息的時候必定要作好消息的版本比較

第三方實現的拒絕策略

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() {
       //省略實現
    }
}複製代碼

能夠看到，當dubbo的工做線程觸發了線程拒絕後，主要作了三個事情，原則就是儘可能讓使用者清楚觸發線程拒絕策略的真實緣由

輸出了一條警告級別的日誌，日誌內容爲線程池的詳細設置參數，以及線程池當前的狀態，還有當前拒絕任務的一些詳細信息。能夠說，這條日誌，使用dubbo的有過生產運維經驗的或多或少是見過的，這個日誌簡直就是日誌打印的典範，其餘的日誌打印的典範還有spring。得益於這麼詳細的日誌，能夠很容易定位到問題所在
輸出當前線程堆棧詳情，這個太有用了，當你經過上面的日誌信息還不能定位問題時，案發現場的dump線程上下文信息就是你發現問題的救命稻草
繼續拋出拒絕執行異常，使本次任務失敗，這個繼承了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中的實現很像JDK中的CallerRunsPolicy，捨不得丟棄任務。不一樣的是，CallerRunsPolicy是直接在調用者線程執行的任務。而 Netty是新建了一個線程來處理的。因此，Netty的實現相較於調用者執行策略的使用面就能夠擴展到支持高效率高性能的場景了。可是也要注意一點，Netty的實現裏，在建立線程時未作任何的判斷約束，也就是說只要系統還有資源就會建立新的線程來處理，直到new不出新的線程了，纔會拋建立線程失敗的異常

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中的策略屬於最大努力執行任務型，當觸發拒絕策略時，在嘗試一分鐘的時間從新將任務塞進任務隊列，當一分鐘超時還沒成功時，就拋出異常

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的拒絕策略實現頗有特色，和其餘的實現都不一樣。他定義了一個拒絕策略鏈，包裝了一個拒絕策略列表，當觸發拒絕策略時，會將策略鏈中的rejectedExecution依次執行一遍