您好,登錄后才能下訂單哦!
今天就跟大家聊聊有關(guān)java線程池ThreadPoolExecutor八種拒絕策略是什么,可能很多人都不太了解,為了讓大家更加了解,小編給大家總結(jié)了以下內(nèi)容,希望大家根據(jù)這篇文章可以有所收獲。
談到j(luò)ava的線程池最熟悉的莫過于ExecutorService接口了,jdk1.5新增的java.util.concurrent包下的這個(gè)api,大大的簡化了多線程代碼的開發(fā)。而不論你用FixedThreadPool還是CachedThreadPool其背后實(shí)現(xiàn)都是ThreadPoolExecutor。ThreadPoolExecutor是一個(gè)典型的緩存池化設(shè)計(jì)的產(chǎn)物,因?yàn)槌刈佑写笮。?dāng)池子體積不夠承載時(shí),就涉及到拒絕策略。JDK中已經(jīng)預(yù)設(shè)了4種線程池拒絕策略,下面結(jié)合場景詳細(xì)聊聊這些策略的使用場景,以及我們還能擴(kuò)展哪些拒絕策略。
池話設(shè)計(jì)應(yīng)該不是一個(gè)新名詞。我們常見的如java線程池、jdbc連接池、redis連接池等就是這類設(shè)計(jì)的代表實(shí)現(xiàn)。這種設(shè)計(jì)會(huì)初始預(yù)設(shè)資源,解決的問題就是抵消每次獲取資源的消耗,如創(chuàng)建線程的開銷,獲取遠(yuǎn)程連接的開銷等。就好比你去食堂打飯,打飯的大媽會(huì)先把飯盛好幾份放那里,你來了就直接拿著飯盒加菜即可,不用再臨時(shí)又盛飯又打菜,效率就高了。除了初始化資源,池化設(shè)計(jì)還包括如下這些特征:池子的初始值、池子的活躍值、池子的最大值等,這些特征可以直接映射到j(luò)ava線程池和數(shù)據(jù)庫連接池的成員屬性中。
和數(shù)據(jù)源連接池不一樣,線程池除了初始大小和池子最大值,還多了一個(gè)阻塞隊(duì)列來緩沖。數(shù)據(jù)源連接池一般請求的連接數(shù)超過連接池的最大值的時(shí)候就會(huì)觸發(fā)拒絕策略,策略一般是阻塞等待設(shè)置的時(shí)間或者直接拋異常。而線程池的觸發(fā)時(shí)機(jī)如下圖:
如圖,想要了解線程池什么時(shí)候觸發(fā)拒絕粗略,需要明確上面三個(gè)參數(shù)的具體含義,是這三個(gè)參數(shù)總體協(xié)調(diào)的結(jié)果,而不是簡單的超過最大線程數(shù)就會(huì)觸發(fā)線程拒絕粗略,當(dāng)提交的任務(wù)數(shù)大于corePoolSize時(shí),會(huì)優(yōu)先放到隊(duì)列緩沖區(qū),只有填滿了緩沖區(qū)后,才會(huì)判斷當(dāng)前運(yùn)行的任務(wù)是否大于maxPoolSize,小于時(shí)會(huì)新建線程處理。大于時(shí)就觸發(fā)了拒絕策略,總結(jié)就是:當(dāng)前提交任務(wù)數(shù)大于(maxPoolSize + queueCapacity)時(shí)就會(huì)觸發(fā)線程池的拒絕策略了。
在分析JDK自帶的線程池拒絕策略前,先看下JDK定義的 拒絕策略接口,如下:
public interface RejectedExecutionHandler { void rejectedExecution(Runnable r, ThreadPoolExecutor executor); }
接口定義很明確,當(dāng)觸發(fā)拒絕策略時(shí),線程池會(huì)調(diào)用你設(shè)置的具體的策略,將當(dāng)前提交的任務(wù)以及線程池實(shí)例本身傳遞給你處理,具體作何處理,不同場景會(huì)有不同的考慮,下面看JDK為我們內(nèi)置了哪些實(shí)現(xiàn):
public static class CallerRunsPolicy implements RejectedExecutionHandler { public CallerRunsPolicy() { } public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { if (!e.isShutdown()) { r.run(); } } }
功能:當(dāng)觸發(fā)拒絕策略時(shí),只要線程池沒有關(guān)閉,就由提交任務(wù)的當(dāng)前線程處理。
使用場景:一般在不允許失敗的、對性能要求不高、并發(fā)量較小的場景下使用,因?yàn)榫€程池一般情況下不會(huì)關(guān)閉,也就是提交的任務(wù)一定會(huì)被運(yùn)行,但是由于是調(diào)用者線程自己執(zhí)行的,當(dāng)多次提交任務(wù)時(shí),就會(huì)阻塞后續(xù)任務(wù)執(zhí)行,性能和效率自然就慢了。
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()); } }
功能:當(dāng)觸發(fā)拒絕策略時(shí),直接拋出拒絕執(zhí)行的異常,中止策略的意思也就是打斷當(dāng)前執(zhí)行流程
使用場景:這個(gè)就沒有特殊的場景了,但是一點(diǎn)要正確處理拋出的異常。ThreadPoolExecutor中默認(rèn)的策略就是AbortPolicy,ExecutorService接口的系列ThreadPoolExecutor因?yàn)槎紱]有顯示的設(shè)置拒絕策略,所以默認(rèn)的都是這個(gè)。但是請注意,ExecutorService中的線程池實(shí)例隊(duì)列都是無界的,也就是說把內(nèi)存撐爆了都不會(huì)觸發(fā)拒絕策略。當(dāng)自己自定義線程池實(shí)例時(shí),使用這個(gè)策略一定要處理好觸發(fā)策略時(shí)拋的異常,因?yàn)樗麜?huì)打斷當(dāng)前的執(zhí)行流程。
public static class DiscardPolicy implements RejectedExecutionHandler { public DiscardPolicy() { } public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { } }
功能:直接靜悄悄的丟棄這個(gè)任務(wù),不觸發(fā)任何動(dòng)作
使用場景:如果你提交的任務(wù)無關(guān)緊要,你就可以使用它 。因?yàn)樗褪莻€(gè)空實(shí)現(xiàn),會(huì)悄無聲息的吞噬你的的任務(wù)。所以這個(gè)策略基本上不用了
public static class DiscardOldestPolicy implements RejectedExecutionHandler { public DiscardOldestPolicy() { } public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { if (!e.isShutdown()) { e.getQueue().poll(); e.execute(r); } } }
功能:如果線程池未關(guān)閉,就彈出隊(duì)列頭部的元素,然后嘗試執(zhí)行
使用場景:這個(gè)策略還是會(huì)丟棄任務(wù),丟棄時(shí)也是毫無聲息,但是特點(diǎn)是丟棄的是老的未執(zhí)行的任務(wù),而且是待執(zhí)行優(yōu)先級較高的任務(wù)。基于這個(gè)特性,我能想到的場景就是,發(fā)布消息,和修改消息,當(dāng)消息發(fā)布出去后,還未執(zhí)行,此時(shí)更新的消息又來了,這個(gè)時(shí)候未執(zhí)行的消息的版本比現(xiàn)在提交的消息版本要低就可以被丟棄了。因?yàn)殛?duì)列中還有可能存在消息版本更低的消息會(huì)排隊(duì)執(zhí)行,所以在真正處理消息的時(shí)候一定要做好消息的版本比較
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() { //省略實(shí)現(xiàn) } }
可以看到,當(dāng)dubbo的工作線程觸發(fā)了線程拒絕后,主要做了三個(gè)事情,原則就是盡量讓使用者清楚觸發(fā)線程拒絕策略的真實(shí)原因
輸出了一條警告級別的日志,日志內(nèi)容為線程池的詳細(xì)設(shè)置參數(shù),以及線程池當(dāng)前的狀態(tài),還有當(dāng)前拒絕任務(wù)的一些詳細(xì)信息??梢哉f,這條日志,使用dubbo的有過生產(chǎn)運(yùn)維經(jīng)驗(yàn)的或多或少是見過的,這個(gè)日志簡直就是日志打印的典范,其他的日志打印的典范還有spring。得益于這么詳細(xì)的日志,可以很容易定位到問題所在
輸出當(dāng)前線程堆棧詳情,這個(gè)太有用了,當(dāng)你通過上面的日志信息還不能定位問題時(shí),案發(fā)現(xiàn)場的dump線程上下文信息就是你發(fā)現(xiàn)問題的救命稻草,這個(gè)可以參考《dubbo線程池耗盡事件-"CyclicBarrier惹的禍"》
繼續(xù)拋出拒絕執(zhí)行異常,使本次任務(wù)失敗,這個(gè)繼承了JDK默認(rèn)拒絕策略的特性
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中的實(shí)現(xiàn)很像JDK中的CallerRunsPolicy,舍不得丟棄任務(wù)。不同的是,CallerRunsPolicy是直接在調(diào)用者線程執(zhí)行的任務(wù)。而 Netty是新建了一個(gè)線程來處理的。所以,Netty的實(shí)現(xiàn)相較于調(diào)用者執(zhí)行策略的使用面就可以擴(kuò)展到支持高效率高性能的場景了。但是也要注意一點(diǎn),Netty的實(shí)現(xiàn)里,在創(chuàng)建線程時(shí)未做任何的判斷約束,也就是說只要系統(tǒng)還有資源就會(huì)創(chuàng)建新的線程來處理,直到new不出新的線程了,才會(huì)拋創(chuàng)建線程失敗的異常
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í)行型策略,當(dāng)觸發(fā)拒絕策略時(shí),會(huì)在次努力一分鐘。重新將任務(wù)塞進(jìn)任務(wù)隊(duì)列,當(dāng)一分鐘超時(shí)還沒成功時(shí),就拋出異常
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的拒絕策略實(shí)現(xiàn)很有特點(diǎn),和其他的實(shí)現(xiàn)都不同。他定義了一個(gè)拒絕策略鏈,包裝了一個(gè)拒絕策略列表,當(dāng)觸發(fā)拒絕策略時(shí),會(huì)將策略鏈中的rejectedExecution依次執(zhí)行一遍
從線程池設(shè)計(jì)思想,以及線程池觸發(fā)拒絕策略的時(shí)機(jī)引出java線程池拒絕策略接口的定義。并輔以JDK內(nèi)置4種以及四個(gè)第三方開源軟件的拒絕策略定義描述了線程池拒絕策略實(shí)現(xiàn)的各種思路和使用場景。
看完上述內(nèi)容,你們對java線程池ThreadPoolExecutor八種拒絕策略是什么有進(jìn)一步的了解嗎?如果還想了解更多知識或者相關(guān)內(nèi)容,請關(guān)注億速云行業(yè)資訊頻道,感謝大家的支持。
免責(zé)聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點(diǎn)不代表本網(wǎng)站立場,如果涉及侵權(quán)請聯(lián)系站長郵箱:is@yisu.com進(jìn)行舉報(bào),并提供相關(guān)證據(jù),一經(jīng)查實(shí),將立刻刪除涉嫌侵權(quán)內(nèi)容。