為什么要用線程池？
諸如 Web 服務(wù)器、數(shù)據(jù)庫服務(wù)器、文件服務(wù)器或郵件服務(wù)器之類的許多服務(wù)器應(yīng)用程序都面臨處理來自某些遠(yuǎn)程來源的大量短小的任務(wù)。請求以某種方式到達(dá)服務(wù)器，這種方式可能是通過網(wǎng)絡(luò)協(xié)議（例如 HTTP、FTP 或 POP）、通過 JMS 隊(duì)列或者可能通過輪詢數(shù)據(jù)庫。不管請求如何到達(dá)，服務(wù)器應(yīng)用程序中經(jīng)常出現(xiàn)的情況是：單個任務(wù)處理的時間很短而請求的數(shù)目卻是巨大的。
只有當(dāng)任務(wù)都是同類型并且相互獨(dú)立時，線程池的性能才能達(dá)到最佳。如果將運(yùn)行時間較長的與運(yùn)行時間較短的任務(wù)混合在一起，那么除非線程池很大，否則將可能造成擁塞，如果提交的任務(wù)依賴于其他任務(wù)，那么除非線程池?zé)o線大，否則將可能造成死鎖。
例如饑餓死鎖：線程池中的任務(wù)需要無限等待一些必須由池中其他任務(wù)才能提供的資源或條件。
ThreadPoolExecutor的通用構(gòu)造函數(shù):（在調(diào)用完構(gòu)造函數(shù)之后可以繼續(xù)定制ThreadPoolExecutor）
public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime, TimeUnit unit,BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory， RejectedExecutionHandler handler){ //...}飽和策略：
ThreadPoolExecutor允許提供一個BlockingQueue來保存等待執(zhí)行的任務(wù)。
當(dāng)有界隊(duì)列被填滿后，飽和策略開始發(fā)揮作用。可以通過調(diào)用setRejectedExecutionHandler來修改。
中止是默認(rèn)的飽和策略，該策略將拋出未檢查的RejectedExecutionException,調(diào)用者可以捕獲這個異常，然后根據(jù)需求編寫自己的處理代碼。
調(diào)用者運(yùn)行策略實(shí)現(xiàn)了一種調(diào)節(jié)機(jī)制，該策略既不會拋棄任務(wù)，也不會拋出異常，而是將某些任務(wù)回退到調(diào)用者，從而降低新任務(wù)的流量。
例如對于WebServer，當(dāng)線程池中的所有線程都被占用，并且工作隊(duì)列被填滿后，下一個任務(wù)在調(diào)用execute時在主線程中執(zhí)行。
由于執(zhí)行任務(wù)需要一定的時間，因此主線程至少在一段時間內(nèi)不能提交任何任務(wù)，從而使得工作者線程有時間來處理完正在執(zhí)行的任務(wù)。
在這期間，主線程不會調(diào)用accept，因此到達(dá)的請求將被保存在TCP層的隊(duì)列中而不是在應(yīng)用程序的隊(duì)列中，如果持續(xù)過載，那么TCP層最終發(fā)現(xiàn)它的請求隊(duì)列被填滿，同樣會開始拋棄請求。
因此當(dāng)服務(wù)器過載時，這種過載會逐漸向外蔓延開來---從線程池到工作隊(duì)列到應(yīng)用程序再到TCP層，最終到達(dá)客戶端，導(dǎo)致服務(wù)器在高負(fù)載下實(shí)現(xiàn)一種平緩的性能降低。
exec.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
當(dāng)工作隊(duì)列被填滿后，沒有預(yù)定于的飽和策略來阻塞execute。而通過Semaphore來現(xiàn)在任務(wù)的到達(dá)率，可以實(shí)現(xiàn)。
/** * 設(shè)置信號量的上界設(shè)置為線程池的大小加上可排隊(duì)任務(wù)的數(shù)量，控制正在執(zhí)行和等待執(zhí)行的任務(wù)數(shù)量。 */public class BoundedExecutor { private final Executor exec; private final Semaphore semaphore; public BoundedExecutor(Executor exec,int bound){ this.exec = exec; this.semaphore = new Semaphore(bound); } public void submitTask(final Runnable task) throws InterruptedException{ semaphore.acquire(); try{ exec.execute(new Runnable(){ public void run(){ try{ task.run(); }finally{ semaphore.release(); } } }); }catch(RejectedExecutionException e){ semaphore.release(); } }}線程工廠
線程池配置信息中可以定制線程工廠，在ThreadFactory中只定義了一個方法newThread，每當(dāng)線程池需要創(chuàng)建一個新線程時都會調(diào)用這個方法。
public interface ThreadFactory{ Thread newThread(Runnable r);}// 示例：將一個特定于線程池的名字傳遞給MyThread的構(gòu)造函數(shù)，從而可以再線程轉(zhuǎn)儲和錯誤日志信息中區(qū)分來自不同線程池的線程。 public class MyThreadFactory implements ThreadFactory{ private final String poolName; public MyThreadFactory(String poolName){ this.poolName = poolName; } public Thread newThread(Runnable runnable){ return new MyThread(runnable,poolName); }}// 示例：為線程指定名字，設(shè)置自定義UncaughtExceptionHandler向Logger中寫入信息及維護(hù)一些統(tǒng)計(jì)信息以及在線程被創(chuàng)建或者終止時把調(diào)試消息寫入日志。public class MyThread extends Thread{ public static final String default_name = "myThread"; private static volatile boolean debugLifecycle = false; private static final AtomicInteger created = new AtomicInteger(); private static final AtomicInteger alive = new AtomicInteger(); private static final Logger log = Logger.getAnonymousLogger(); public MyThread(Runnable runnable){ this(runnable,default_name); } public MyThread(Runnable runnable, String defaultName) { super(runnable,defaultName + "-" + created.incrementAndGet()); setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() { @Override public void uncaughtException(Thread t, Throwable e) { log.log(Level.SEVERE,"uncaught in thread " + t.getName(), e); } }); } public void run(){ boolean debug = debugLifecycle; if(debug){ log.log(Level.FINE,"created " + getName()); } try{ alive.incrementAndGet(); super.run(); }finally{ alive.decrementAndGet(); if(debug){ log.log(Level.FINE,"Exiting " + getName()); } } }}擴(kuò)展ThreadPoolExecutor
在線程池完成關(guān)閉操作時調(diào)用terminated,也就是在所有任務(wù)都已經(jīng)完成并且所有工作者線程也已經(jīng)關(guān)閉后。terminated可以用來釋放Executor在其生命周期里分配的各種資源，此外還可以執(zhí)行發(fā)送通知、記錄日志或者收集finalize統(tǒng)計(jì)信息等操作。
示例：給線程池添加統(tǒng)計(jì)信息
/** * TimingThreadPool中給出了一個自定義的線程池，通過beforeExecute、afterExecute、terminated等方法來添加日志記錄和統(tǒng)計(jì)信息收集。 * 為了測量任務(wù)的運(yùn)行時間，beforeExecute必須記錄開始時間并把它保存到一個afterExecute可用訪問的地方。 * 因?yàn)檫@些方法將在執(zhí)行任務(wù)的線程中調(diào)用，因此beforeExecute可以把值保存到一個ThreadLocal變量中。然后由afterExecute來取。 * 在TimingThreadPool中使用了兩個AtomicLong變量，分別用于記錄已處理的任務(wù)和總的處理時間，并通過包含平均任務(wù)時間的日志消息。 */public class TimingThreadPool extends ThreadPoolExecutor{ public TimingThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) { super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue); } private final ThreadLocal<Long> startTime = new ThreadLocal<Long>(); private final Logger log = Logger.getLogger("TimingThreadPool"); private final AtomicLong numTasks = new AtomicLong(); private final AtomicLong totalTime = new AtomicLong(); protected void beforeExecute(Thread t,Runnable r){ super.beforeExecute(t, r); log.fine(String.format("Thread %s: start %s", t,r)); startTime.set(System.nanoTime()); } protected void afterExecute(Throwable t,Runnable r){ try{ long endTime = System.nanoTime(); long taskTime = endTime - startTime.get(); numTasks.incrementAndGet(); totalTime.addAndGet(taskTime); log.fine(String.format("Thread %s: end %s, time=%dns", t,r,taskTime)); }finally{ super.afterExecute(r, t); } } protected void terminated(){ try{ log.info(String.format("Terminated: avg time=%dns", totalTime.get()/numTasks.get())); }finally{ super.terminated(); } }}#筆記內(nèi)容參考 《java并發(fā)編程實(shí)戰(zhàn)》
總結(jié)
以上就是這篇文章的全部內(nèi)容了，希望本文的內(nèi)容對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，如果有疑問大家可以留言交流，謝謝大家對VeVb武林網(wǎng)的支持。
新聞熱點(diǎn)
疑難解答
圖片精選
