|
Java線程池使用說(shuō)明 一簡(jiǎn)介線程的使用在java中占有極其重要的地位,，在jdk1.4極其之前的jdk版本中,，關(guān)于線程池的使用是極其簡(jiǎn)陋的。在jdk1.5之后這一情況有了很大的改觀,。Jdk1.5之后加入了java.util.concurrent包,，這個(gè)包中主要介紹java中線程以及線程池的使用。為我們?cè)陂_發(fā)中處理線程的問(wèn)題提供了非常大的幫助,。 二：線程池線程池的作用： 線程池作用就是限制系統(tǒng)中執(zhí)行線程的數(shù)量,。 為什么要用線程池: 1.減少了創(chuàng)建和銷毀線程的次數(shù),，每個(gè)工作線程都可以被重復(fù)利用，可執(zhí)行多個(gè)任務(wù),。 2.可以根據(jù)系統(tǒng)的承受能力,，調(diào)整線程池中工作線線程的數(shù)目，防止因?yàn)橄倪^(guò)多的內(nèi)存,，而把服務(wù)器累趴下(每個(gè)線程需要大約1MB內(nèi)存,，線程開的越多，消耗的內(nèi)存也就越大,，最后死機(jī)),。 Java里面線程池的頂級(jí)接口是Executor，但是嚴(yán)格意義上講Executor并不是一個(gè)線程池,，而只是一個(gè)執(zhí)行線程的工具,。真正的線程池接口是ExecutorService。 比較重要的幾個(gè)類：
要配置一個(gè)線程池是比較復(fù)雜的,，尤其是對(duì)于線程池的原理不是很清楚的情況下,，很有可能配置的線程池不是較優(yōu)的，因此在Executors類里面提供了一些靜態(tài)工廠,，生成一些常用的線程池,。 1. newSingleThreadExecutor 創(chuàng)建一個(gè)單線程的線程池。這個(gè)線程池只有一個(gè)線程在工作,，也就是相當(dāng)于單線程串行執(zhí)行所有任務(wù),。如果這個(gè)唯一的線程因?yàn)楫惓＝Y(jié)束，那么會(huì)有一個(gè)新的線程來(lái)替代它,。此線程池保證所有任務(wù)的執(zhí)行順序按照任務(wù)的提交順序執(zhí)行,。 2.newFixedThreadPool 創(chuàng)建固定大小的線程池。每次提交一個(gè)任務(wù)就創(chuàng)建一個(gè)線程,，直到線程達(dá)到線程池的最大大小,。線程池的大小一旦達(dá)到最大值就會(huì)保持不變，如果某個(gè)線程因?yàn)閳?zhí)行異常而結(jié)束,，那么線程池會(huì)補(bǔ)充一個(gè)新線程,。 3. newCachedThreadPool 創(chuàng)建一個(gè)可緩存的線程池。如果線程池的大小超過(guò)了處理任務(wù)所需要的線程,， 那么就會(huì)回收部分空閑（60秒不執(zhí)行任務(wù)）的線程,，當(dāng)任務(wù)數(shù)增加時(shí)，此線程池又可以智能的添加新線程來(lái)處理任務(wù),。此線程池不會(huì)對(duì)線程池大小做限制,，線程池大小完全依賴于操作系統(tǒng)（或者說(shuō)JVM）能夠創(chuàng)建的最大線程大小。 4.newScheduledThreadPool 創(chuàng)建一個(gè)大小無(wú)限的線程池,。此線程池支持定時(shí)以及周期性執(zhí)行任務(wù)的需求,。 實(shí)例 1：newSingleThreadExecutor MyThread.java
TestSingleThreadExecutor.java
輸出結(jié)果
2newFixedThreadPool TestFixedThreadPool.Java
輸出結(jié)果
3 newCachedThreadPool TestCachedThreadPool.java
輸出結(jié)果：
4newScheduledThreadPool TestScheduledThreadPoolExecutor.java
輸出結(jié)果
三：ThreadPoolExecutor詳解ThreadPoolExecutor的完整構(gòu)造方法的簽名是：ThreadPoolExecutor corePoolSize - 池中所保存的線程數(shù)，包括空閑線程,。 maximumPoolSize-池中允許的最大線程數(shù),。 keepAliveTime - 當(dāng)線程數(shù)大于核心時(shí)，此為終止前多余的空閑線程等待新任務(wù)的最長(zhǎng)時(shí)間,。 unit - keepAliveTime 參數(shù)的時(shí)間單位,。 workQueue - 執(zhí)行前用于保持任務(wù)的隊(duì)列。此隊(duì)列僅保持由 execute方法提交的 Runnable任務(wù),。 threadFactory - 執(zhí)行程序創(chuàng)建新線程時(shí)使用的工廠,。 handler - 由于超出線程范圍和隊(duì)列容量而使執(zhí)行被阻塞時(shí)所使用的處理程序。 ThreadPoolExecutor是Executors類的底層實(shí)現(xiàn),。 在JDK幫助文檔中,，有如此一段話： “強(qiáng)烈建議程序員使用較為方便的 它們均為大多數(shù)使用場(chǎng)景預(yù)定義了設(shè)置,。” 下面介紹一下幾個(gè)類的源碼： ExecutorService newFixedThreadPool (int nThreads):固定大小線程池。 可以看到，corePoolSize和maximumPoolSize的大小是一樣的（實(shí)際上,，后面會(huì)介紹,，如果使用無(wú)界queue的話maximumPoolSize參數(shù)是沒(méi)有意義的），keepAliveTime和unit的設(shè)值表名什么,？-就是該實(shí)現(xiàn)不想keep alive,！最后的BlockingQueue選擇了LinkedBlockingQueue,，該queue有一個(gè)特點(diǎn),，他是無(wú)界的。
ExecutorService newSingleThreadExecutor()：?jiǎn)尉€程
ExecutorService newCachedThreadPool()：無(wú)界線程池,，可以進(jìn)行自動(dòng)線程回收 這個(gè)實(shí)現(xiàn)就有意思了,。首先是無(wú)界的線程池，所以我們可以發(fā)現(xiàn)maximumPoolSize為big big,。其次BlockingQueue的選擇上使用SynchronousQueue,。可能對(duì)于該BlockingQueue有些陌生,，簡(jiǎn)單說(shuō)：該QUEUE中,，每個(gè)插入操作必須等待另一個(gè)線程的對(duì)應(yīng)移除操作。
先從BlockingQueue<Runnable> workQueue這個(gè)入?yún)㈤_始說(shuō)起,。在JDK中,，其實(shí)已經(jīng)說(shuō)得很清楚了，一共有三種類型的queue,。 所有BlockingQueue 都可用于傳輸和保持提交的任務(wù),。可以使用此隊(duì)列與池大小進(jìn)行交互： 如果運(yùn)行的線程少于 corePoolSize,，則 Executor始終首選添加新的線程,，而不進(jìn)行排隊(duì)。（如果當(dāng)前運(yùn)行的線程小于corePoolSize,，則任務(wù)根本不會(huì)存放,，添加到queue中，而是直接抄家伙（thread）開始運(yùn)行） 如果運(yùn)行的線程等于或多于 corePoolSize,，則 Executor始終首選將請(qǐng)求加入隊(duì)列,，而不添加新的線程。 如果無(wú)法將請(qǐng)求加入隊(duì)列,，則創(chuàng)建新的線程,，除非創(chuàng)建此線程超出 maximumPoolSize，在這種情況下,，任務(wù)將被拒絕,。 queue上的三種類型。
排隊(duì)有三種通用策略： 直接提交。工作隊(duì)列的默認(rèn)選項(xiàng)是 SynchronousQueue,，它將任務(wù)直接提交給線程而不保持它們,。在此，如果不存在可用于立即運(yùn)行任務(wù)的線程,，則試圖把任務(wù)加入隊(duì)列將失敗,，因此會(huì)構(gòu)造一個(gè)新的線程。此策略可以避免在處理可能具有內(nèi)部依賴性的請(qǐng)求集時(shí)出現(xiàn)鎖,。直接提交通常要求無(wú)界 maximumPoolSizes 以避免拒絕新提交的任務(wù),。當(dāng)命令以超過(guò)隊(duì)列所能處理的平均數(shù)連續(xù)到達(dá)時(shí)，此策略允許無(wú)界線程具有增長(zhǎng)的可能性,。 無(wú)界隊(duì)列,。使用無(wú)界隊(duì)列（例如，不具有預(yù)定義容量的 LinkedBlockingQueue）將導(dǎo)致在所有 corePoolSize 線程都忙時(shí)新任務(wù)在隊(duì)列中等待,。這樣,，創(chuàng)建的線程就不會(huì)超過(guò) corePoolSize。（因此,，maximumPoolSize的值也就無(wú)效了,。）當(dāng)每個(gè)任務(wù)完全獨(dú)立于其他任務(wù)，即任務(wù)執(zhí)行互不影響時(shí),，適合于使用無(wú)界隊(duì)列,；例如，在 Web頁(yè)服務(wù)器中,。這種排隊(duì)可用于處理瞬態(tài)突發(fā)請(qǐng)求,，當(dāng)命令以超過(guò)隊(duì)列所能處理的平均數(shù)連續(xù)到達(dá)時(shí)，此策略允許無(wú)界線程具有增長(zhǎng)的可能性,。 有界隊(duì)列,。當(dāng)使用有限的 maximumPoolSizes時(shí)，有界隊(duì)列（如 ArrayBlockingQueue）有助于防止資源耗盡,，但是可能較難調(diào)整和控制,。隊(duì)列大小和最大池大小可能需要相互折衷：使用大型隊(duì)列和小型池可以最大限度地降低 CPU 使用率、操作系統(tǒng)資源和上下文切換開銷,，但是可能導(dǎo)致人工降低吞吐量,。如果任務(wù)頻繁阻塞（例如，如果它們是 I/O邊界）,，則系統(tǒng)可能為超過(guò)您許可的更多線程安排時(shí)間,。使用小型隊(duì)列通常要求較大的池大小，CPU使用率較高,，但是可能遇到不可接受的調(diào)度開銷,，這樣也會(huì)降低吞吐量,。 BlockingQueue的選擇。 例子一：使用直接提交策略,，也即SynchronousQueue,。 首先SynchronousQueue是無(wú)界的，也就是說(shuō)他存數(shù)任務(wù)的能力是沒(méi)有限制的,，但是由于該Queue本身的特性,，在某次添加元素后必須等待其他線程取走后才能繼續(xù)添加。在這里不是核心線程便是新創(chuàng)建的線程,，但是我們?cè)囅胍粯酉?，下面的?chǎng)景。 我們使用一下參數(shù)構(gòu)造ThreadPoolExecutor： 1. new ThreadPoolExecutor( 2. 2, 3, 30, TimeUnit.SECONDS, 3. new SynchronousQueue<Runnable>(), 4. new RecorderThreadFactory("CookieRecorderPool"),
new ThreadPoolExecutor( 2, 3, 30, TimeUnit.SECONDS, new SynchronousQueue<Runnable>(), new RecorderThreadFactory("CookieRecorderPool"), new ThreadPoolExecutor.CallerRunsPolicy()); 當(dāng)核心線程已經(jīng)有2個(gè)正在運(yùn)行.
所以在使用SynchronousQueue通常要求maximumPoolSize是無(wú)界的,，這樣就可以避免上述情況發(fā)生（如果希望限制就直接使用有界隊(duì)列）,。對(duì)于使用SynchronousQueue的作用jdk中寫的很清楚：此策略可以避免在處理可能具有內(nèi)部依賴性的請(qǐng)求集時(shí)出現(xiàn)鎖,。 什么意思？如果你的任務(wù)A1,，A2有內(nèi)部關(guān)聯(lián),，A1需要先運(yùn)行，那么先提交A1,，再提交A2,，當(dāng)使用SynchronousQueue我們可以保證，A1必定先被執(zhí)行,，在A1么有被執(zhí)行前,，A2不可能添加入queue中。 例子二：使用無(wú)界隊(duì)列策略,，即LinkedBlockingQueue 這個(gè)就拿newFixedThreadPool來(lái)說(shuō),，根據(jù)前文提到的規(guī)則： 如果運(yùn)行的線程少于 corePoolSize，則 Executor 始終首選添加新的線程,，而不進(jìn)行排隊(duì),。那么當(dāng)任務(wù)繼續(xù)增加，會(huì)發(fā)生什么呢,？ 如果運(yùn)行的線程等于或多于 corePoolSize,，則 Executor 始終首選將請(qǐng)求加入隊(duì)列，而不添加新的線程,。OK,，此時(shí)任務(wù)變加入隊(duì)列之中了，那什么時(shí)候才會(huì)添加新線程呢,？ 如果無(wú)法將請(qǐng)求加入隊(duì)列,，則創(chuàng)建新的線程，除非創(chuàng)建此線程超出 maximumPoolSize,，在這種情況下,，任務(wù)將被拒絕。這里就很有意思了,，可能會(huì)出現(xiàn)無(wú)法加入隊(duì)列嗎,？不像SynchronousQueue那樣有其自身的特點(diǎn)，對(duì)于無(wú)界隊(duì)列來(lái)說(shuō),，總是可以加入的（資源耗盡,，當(dāng)然另當(dāng)別論）。換句說(shuō),，永遠(yuǎn)也不會(huì)觸發(fā)產(chǎn)生新的線程,！corePoolSize大小的線程數(shù)會(huì)一直運(yùn)行，忙完當(dāng)前的,，就從隊(duì)列中拿任務(wù)開始運(yùn)行,。所以要防止任務(wù)瘋長(zhǎng),，比如任務(wù)運(yùn)行的實(shí)行比較長(zhǎng)，而添加任務(wù)的速度遠(yuǎn)遠(yuǎn)超過(guò)處理任務(wù)的時(shí)間,，而且還不斷增加,，不一會(huì)兒就爆了。 例子三：有界隊(duì)列,，使用ArrayBlockingQueue,。 這個(gè)是最為復(fù)雜的使用，所以JDK不推薦使用也有些道理,。與上面的相比,，最大的特點(diǎn)便是可以防止資源耗盡的情況發(fā)生。 舉例來(lái)說(shuō),，請(qǐng)看如下構(gòu)造方法： 1. new ThreadPoolExecutor( 2. 2, 4, 30, TimeUnit.SECONDS, 3. new ArrayBlockingQueue<Runnable>(2), 4. new RecorderThreadFactory("CookieRecorderPool"), 5. new ThreadPoolExecutor.CallerRunsPolicy()); new ThreadPoolExecutor( 2, 4, 30, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(2), new RecorderThreadFactory("CookieRecorderPool"), new ThreadPoolExecutor.CallerRunsPolicy()); 假設(shè),，所有的任務(wù)都永遠(yuǎn)無(wú)法執(zhí)行完。 對(duì)于首先來(lái)的A,B來(lái)說(shuō)直接運(yùn)行,，接下來(lái),，如果來(lái)了C,D，他們會(huì)被放到queue中,，如果接下來(lái)再來(lái)E,F,，則增加線程運(yùn)行E，F,。但是如果再來(lái)任務(wù)，隊(duì)列無(wú)法再接受了,，線程數(shù)也到達(dá)最大的限制了,，所以就會(huì)使用拒絕策略來(lái)處理。 keepAliveTime jdk中的解釋是：當(dāng)線程數(shù)大于核心時(shí),，此為終止前多余的空閑線程等待新任務(wù)的最長(zhǎng)時(shí)間,。 有點(diǎn)拗口，其實(shí)這個(gè)不難理解,，在使用了“池”的應(yīng)用中,，大多都有類似的參數(shù)需要配置。比如數(shù)據(jù)庫(kù)連接池,，DBCP中的maxIdle,，minIdle參數(shù)。 什么意思,？接著上面的解釋,，后來(lái)向老板派來(lái)的工人始終是“借來(lái)的”，俗話說(shuō)“有借就有還”,，但這里的問(wèn)題就是什么時(shí)候還了,，如果借來(lái)的工人剛完成一個(gè)任務(wù)就還回去,，后來(lái)發(fā)現(xiàn)任務(wù)還有，那豈不是又要去借,？這一來(lái)一往,，老板肯定頭也大死了。
合理的策略：既然借了,，那就多借一會(huì)兒,。直到“某一段”時(shí)間后，發(fā)現(xiàn)再也用不到這些工人時(shí),，便可以還回去了,。這里的某一段時(shí)間便是keepAliveTime的含義，TimeUnit為keepAliveTime值的度量,。
RejectedExecutionHandler 另一種情況便是,，即使向老板借了工人，但是任務(wù)還是繼續(xù)過(guò)來(lái),，還是忙不過(guò)來(lái),，這時(shí)整個(gè)隊(duì)伍只好拒絕接受了。 RejectedExecutionHandler接口提供了對(duì)于拒絕任務(wù)的處理的自定方法的機(jī)會(huì),。在ThreadPoolExecutor中已經(jīng)默認(rèn)包含了4中策略,，因?yàn)樵创a非常簡(jiǎn)單，這里直接貼出來(lái),。 CallerRunsPolicy：線程調(diào)用運(yùn)行該任務(wù)的 execute 本身,。此策略提供簡(jiǎn)單的反饋控制機(jī)制，能夠減緩新任務(wù)的提交速度,。 1. public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { 2. if (!e.isShutdown()) { 3. r.run(); 4. } 5. } public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { if (!e.isShutdown()) { r.run(); } } 這個(gè)策略顯然不想放棄執(zhí)行任務(wù),。但是由于池中已經(jīng)沒(méi)有任何資源了，那么就直接使用調(diào)用該execute的線程本身來(lái)執(zhí)行,。 AbortPolicy：處理程序遭到拒絕將拋出運(yùn)行時(shí)RejectedExecutionException 1. public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { 2. throw new RejectedExecutionException(); 3. } public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { throw new RejectedExecutionException(); } 這種策略直接拋出異常,，丟棄任務(wù)。 DiscardPolicy：不能執(zhí)行的任務(wù)將被刪除 1. public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { 2. } public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { } 這種策略和AbortPolicy幾乎一樣,，也是丟棄任務(wù),，只不過(guò)他不拋出異常。 DiscardOldestPolicy：如果執(zhí)行程序尚未關(guān)閉,，則位于工作隊(duì)列頭部的任務(wù)將被刪除,，然后重試執(zhí)行程序（如果再次失敗，則重復(fù)此過(guò)程） 1. public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { 2. if (!e.isShutdown()) { 3. e.getQueue().poll(); 4. e.execute(r); 5. }
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { if (!e.isShutdown()) { e.getQueue().poll(); e.execute(r); } } 該策略就稍微復(fù)雜一些,，在pool沒(méi)有關(guān)閉的前提下首先丟掉緩存在隊(duì)列中的最早的任務(wù),，然后重新嘗試運(yùn)行該任務(wù)。這個(gè)策略需要適當(dāng)小心,。 設(shè)想:如果其他線程都還在運(yùn)行,，那么新來(lái)任務(wù)踢掉舊任務(wù),，緩存在queue中，再來(lái)一個(gè)任務(wù)又會(huì)踢掉queue中最老任務(wù),。 總結(jié)： keepAliveTime和maximumPoolSize及BlockingQueue的類型均有關(guān)系,。如果BlockingQueue是無(wú)界的，那么永遠(yuǎn)不會(huì)觸發(fā)maximumPoolSize,，自然keepAliveTime也就沒(méi)有了意義,。 反之，如果核心數(shù)較小,，有界BlockingQueue數(shù)值又較小,，同時(shí)keepAliveTime又設(shè)的很小，如果任務(wù)頻繁,，那么系統(tǒng)就會(huì)頻繁的申請(qǐng)回收線程,。
public static ExecutorService newFixedThreadPool(int nThreads) { return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>()); } 原文鏈接：http://blog.csdn.net/sd0902/article/details/8395677 |
|
|
