ThreadFactory

線(xiàn)程池中的線(xiàn)程從哪里來(lái)呢？就是ThreadFoctory

public interface ThreadFactory { Thread newThread(Runnable r);}

Threadfactory里面有個(gè)接口，當(dāng)線(xiàn)程池中需要?jiǎng)?chuàng)建線(xiàn)程就會(huì)調(diào)用該方法，也可以自定義線(xiàn)程工廠

public class ThreadfactoryText { public static void main(String[] args) {Runnable runnable=new Runnable() { @Override public void run() {int num=new Random().nextInt(10);System.out.println(Thread.currentThread().getId()+'--'+System.currentTimeMillis()+'--睡眠'+num);try { TimeUnit.SECONDS.sleep(num);} catch (InterruptedException e) { e.printStackTrace();} }};//創(chuàng)建線(xiàn)程池 使用自定義線(xiàn)程工廠 采用默認(rèn)的拒絕策略ExecutorService executorService=new ThreadPoolExecutor(5, 5, 0, TimeUnit.SECONDS, new SynchronousQueue<>(), new ThreadFactory() { @Override public Thread newThread(Runnable r) {Thread t=new Thread(r);t.setDaemon(true);//設(shè)置為守護(hù)線(xiàn)程，當(dāng)主線(xiàn)程運(yùn)行結(jié)束，線(xiàn)程池中線(xiàn)程也會(huì)被釋放System.out.println('創(chuàng)建了線(xiàn)程'+t);return t; }});//提交五個(gè)任務(wù)for (int i = 0; i < 5; i++) { executorService.submit(runnable);} }}

當(dāng)線(xiàn)程提交超過(guò)五個(gè)任務(wù)時(shí)，線(xiàn)程池會(huì)默認(rèn)拋出異常

監(jiān)控線(xiàn)程池

ThreadPoolExcutor提供了一組方法用于監(jiān)控線(xiàn)程池

int getActiveCount()//獲得線(xiàn)程池只當(dāng)前的獲得線(xiàn)程數(shù)量long getCompletedTaskCount()//返回線(xiàn)程池完成任務(wù)數(shù)量int getCorePoolSize()//線(xiàn)程池中核心任務(wù)數(shù)量int getLargestPoolSize() //返回線(xiàn)程池中曾經(jīng)達(dá)到線(xiàn)程的最大數(shù)int getMaximumPoolSize()//返回線(xiàn)程池的最大容量int getPoolSize()//返回線(xiàn)程大小BlockingQueue<Runnable> getQueue()//返回阻塞隊(duì)列l(wèi)ong getTaskCount()//返回線(xiàn)程池收到任務(wù)總數(shù)

public class Text { public static void main(String[] args) throws InterruptedException {Runnable runnable = new Runnable() { @Override public void run() {System.out.println(Thread.currentThread().getId() + '線(xiàn)程開(kāi)始執(zhí)行--' + System.currentTimeMillis());try { Thread.sleep(10000);} catch (InterruptedException e) { e.printStackTrace();} }};//創(chuàng)建線(xiàn)程池 使用默認(rèn)線(xiàn)程工廠 有界隊(duì)列 采用DiscardPolicy策略ThreadPoolExecutor executorService = new ThreadPoolExecutor(2, 5, 0, TimeUnit.SECONDS, new ArrayBlockingQueue<>(5),Executors.defaultThreadFactory(),new ThreadPoolExecutor.DiscardPolicy());//提交五個(gè)任務(wù)for (int i = 0; i < 30; i++) { executorService.submit(runnable); System.out.println('當(dāng)前線(xiàn)程核心線(xiàn)程數(shù)'+executorService.getCorePoolSize()+',最大線(xiàn)程數(shù)：'+executorService.getMaximumPoolSize()+',當(dāng)前線(xiàn)程池大小：'+executorService.getPoolSize()+'活動(dòng)線(xiàn)程數(shù)：'+executorService.getActiveCount()+',收到任務(wù)：'+executorService.getTaskCount()+'完成任務(wù)數(shù)：'+executorService.getCompletedTaskCount()+'等待任務(wù)數(shù)：'+executorService.getQueue().size()); TimeUnit.MILLISECONDS.sleep(500);}System.out.println('-------------------');while (executorService.getActiveCount()>=0)//繼續(xù)對(duì)線(xiàn)程池進(jìn)行檢測(cè){ System.out.println('當(dāng)前線(xiàn)程核心線(xiàn)程數(shù)'+executorService.getCorePoolSize()+',最大線(xiàn)程數(shù)：'+executorService.getMaximumPoolSize()+',當(dāng)前線(xiàn)程池大小：'+executorService.getPoolSize()+'活動(dòng)線(xiàn)程數(shù)：'+executorService.getActiveCount()+',收到任務(wù)：'+executorService.getTaskCount()+'完成任務(wù)數(shù)：'+executorService.getCompletedTaskCount()+'等待任務(wù)數(shù)：'+executorService.getQueue().size()); Thread.sleep(1000);//每1秒檢測(cè)一次} }}

當(dāng)線(xiàn)程池大小達(dá)到了核心線(xiàn)程數(shù)，線(xiàn)程會(huì)被放在等待隊(duì)列。當(dāng)線(xiàn)程池等待隊(duì)列已滿(mǎn)會(huì)開(kāi)啟新的線(xiàn)程。當(dāng)當(dāng)前線(xiàn)程大小達(dá)到最大線(xiàn)程數(shù)，等待隊(duì)列也滿(mǎn)了，再提交的話(huà)會(huì)執(zhí)行DiscardPolicy策略，直接丟棄這個(gè)無(wú)法處理的任務(wù)，最后30個(gè)任務(wù)只剩下15個(gè)了。

原理如圖：

擴(kuò)展線(xiàn)程池

有時(shí)候需要對(duì)線(xiàn)程池進(jìn)行擴(kuò)展，如在監(jiān)控每個(gè)任務(wù)開(kāi)始和結(jié)束時(shí)間，或者自定義其他增強(qiáng)功能。

ThreadPoolExecutor線(xiàn)程池提供了兩個(gè)方法：

protected void beforeExecute(Thread t, Runnable r) { }protected void afterExecute(Runnable r, Throwable t) { }

線(xiàn)程池執(zhí)行某個(gè)任務(wù)前會(huì)執(zhí)行beforeExecute()方法，執(zhí)行后會(huì)調(diào)用afterExecute()方法

查看ThreadPoolExecutor源碼，在該類(lèi)中定義了一個(gè)內(nèi)部類(lèi)Worker，ThreadPoolExecutor線(xiàn)程池的工作線(xiàn)程就是Worker類(lèi)的實(shí)例，Worker實(shí)例在執(zhí)行時(shí)會(huì)調(diào)用beforeExecute與afterExecute方法。

public void run() { runWorker(this);}final void runWorker(Worker w) {try { beforeExecute(wt, task); try {task.run();afterExecute(task, null); } catch (Throwable ex) {afterExecute(task, ex);throw ex; }} finally { task = null; w.completedTasks++; w.unlock();} } }

部分代碼已省略，線(xiàn)程執(zhí)行前會(huì)調(diào)用beforeExecute，執(zhí)行后會(huì)調(diào)用afterExecute方法。

擴(kuò)展線(xiàn)程池示例

package com;import java.util.concurrent.ExecutorService;import java.util.concurrent.LinkedBlockingDeque;import java.util.concurrent.ThreadPoolExecutor;import java.util.concurrent.TimeUnit;public class Text07 { public static void main(String[] args) {//定義擴(kuò)展線(xiàn)程池 定義線(xiàn)程池類(lèi)繼承ThreadPoolExecutor，然后重寫(xiě)其他方法ExecutorService threadPoolExecutor= new ThreadPoolExecutor(5,5,0, TimeUnit.SECONDS,new LinkedBlockingDeque<>()){ //在內(nèi)部類(lèi)重寫(xiě)開(kāi)始方法 @Override protected void beforeExecute(Thread t, Runnable r) { System.out.println(t.getId()+'線(xiàn)程準(zhǔn)備執(zhí)行任務(wù)'+((Mytask)r).name); } //在內(nèi)部類(lèi)重寫(xiě)結(jié)束方法 @Override protected void afterExecute(Runnable r, Throwable t) { System.out.println(((Mytask)r).name+'執(zhí)行完成'); } //線(xiàn)程池退出 @Override protected void terminated() { System.out.println('線(xiàn)程池退出'); } };for (int i = 0; i < 5; i++) { Mytask mytask=new Mytask('Thread'+i); threadPoolExecutor.execute(mytask);} } private static class Mytask implements Runnable {private String name;public Mytask(String name){ this.name=name;}@Overridepublic void run() { System.out.println(name+'正在被執(zhí)行'+Thread.currentThread().getId()); try {Thread.sleep(1000);//模擬任務(wù)時(shí)長(zhǎng) } catch (InterruptedException e) {e.printStackTrace(); }} }}

優(yōu)化線(xiàn)程池大小

線(xiàn)程池大小對(duì)系統(tǒng)性能有一定影響，過(guò)大或者過(guò)小都無(wú)法方法發(fā)揮系統(tǒng)最佳性能，不需要非常精確，只要避免極大或者極小就可以了，一般來(lái)說(shuō)線(xiàn)程池大小大姚考慮CPU數(shù)量

線(xiàn)程池大小=CPU數(shù)量 * 目標(biāo)CPU使用率*（1+等待時(shí)間與計(jì)算時(shí)間的比）

線(xiàn)程池死鎖

如果線(xiàn)程池執(zhí)行中，任務(wù)A在執(zhí)行過(guò)程中提交了任務(wù)B，任務(wù)B添加到線(xiàn)程池中的等待隊(duì)列，如果A的結(jié)束需要B的執(zhí)行結(jié)果，而B(niǎo)線(xiàn)程需要等待A線(xiàn)程執(zhí)行完畢，就可能會(huì)使其他所有工作線(xiàn)程都處于等待狀態(tài)，待這些任務(wù)在阻塞隊(duì)列中執(zhí)行。線(xiàn)程池中沒(méi)有可以對(duì)阻塞隊(duì)列進(jìn)行處理的線(xiàn)程，就會(huì)一直等待下去照成死鎖。

適合給線(xiàn)程池提交相互獨(dú)立的任務(wù)，而不是彼此依賴(lài)的任務(wù)，對(duì)于彼此依賴(lài)的任務(wù)，可以考慮分別提交給不同的線(xiàn)程池來(lái)處理。

線(xiàn)程池異常信息捕獲

import java.util.concurrent.ExecutorService;import java.util.concurrent.SynchronousQueue;import java.util.concurrent.ThreadPoolExecutor;import java.util.concurrent.TimeUnit;public class Text09 { public static void main(String[] args) {//創(chuàng)建線(xiàn)程池ExecutorService executorService=new ThreadPoolExecutor(5,5,0, TimeUnit.SECONDS,new SynchronousQueue<>());//向線(xiàn)程池中添加兩個(gè)數(shù)相處計(jì)算的任務(wù)for (int i = 0; i <5 ; i++) { executorService.submit(new Text(10,i));} } private static class Text implements Runnable {private int x;private int y;public Text(int x,int y){ this.x=x; this.y=y;}@Overridepublic void run() { System.out.println(Thread.currentThread().getName()+'線(xiàn)程x/y結(jié)果的為'+x+'/'+y+'='+(x/y));} }}

可以看到只有四條結(jié)果，實(shí)際向線(xiàn)程池提交了五個(gè)任務(wù)，但是當(dāng)i==0時(shí)，產(chǎn)生了算術(shù)異常，線(xiàn)程池把該異常吃掉了，導(dǎo)致我們對(duì)該異常一無(wú)所知

解決辦法：

1.把submit改為execute

2.對(duì)線(xiàn)程池進(jìn)行擴(kuò)展，對(duì)submit進(jìn)行包裝

package com;import java.util.concurrent.*;public class Text09 { public static void main(String[] args) {//創(chuàng)建線(xiàn)程池 使用自定義的線(xiàn)程池ExecutorService executorService=new TranceThreadPoorExcuter(5,5,0, TimeUnit.SECONDS,new SynchronousQueue<>());//向線(xiàn)程池中添加兩個(gè)數(shù)相處計(jì)算的任務(wù)for (int i = 0; i <5 ; i++) { executorService.submit(new Text(10,i));} } public static class Text implements Runnable {public int x;public int y;public Text(int x,int y){ this.x=x; this.y=y;}@Overridepublic void run() { System.out.println(Thread.currentThread().getName()+'線(xiàn)程x/y結(jié)果的為'+x+'/'+y+'='+(x/y));} } //自定義線(xiàn)程池類(lèi) 對(duì)TranceThreadPoorExcuter進(jìn)行擴(kuò)展 private static class TranceThreadPoorExcuter extends ThreadPoolExecutor {public TranceThreadPoorExcuter(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) { super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);}//定義一個(gè)方法用于傳入兩個(gè)參數(shù) 第一個(gè)是要接受的任務(wù) 第二個(gè)是Exceptionpublic Runnable warp(Runnable r,Exception e){ return new Runnable() {@Overridepublic void run() { try {r.run(); } catch (Exception e1) {e.printStackTrace();throw e1; }} };}//重寫(xiě)submit方法@Overridepublic Future<?> submit(Runnable task) { return super.submit(warp(task,new Exception('客戶(hù)跟蹤異常')));}//還可以重寫(xiě)excute方法 }}

此方法使用了自定義的線(xiàn)程池，重寫(xiě)線(xiàn)程池中的submit方法，在submit方法中，把要傳入的任務(wù)參數(shù)帶一個(gè)捕獲異常信息的功能就可以捕獲線(xiàn)程池異常。

到此這篇關(guān)于Java利用線(xiàn)程工廠監(jiān)控線(xiàn)程池的實(shí)現(xiàn)示例的文章就介紹到這了,更多相關(guān)Java 線(xiàn)程工廠監(jiān)控線(xiàn)程池內(nèi)容請(qǐng)搜索好吧啦網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持好吧啦網(wǎng)！