ThreadPoolExecutor
創(chuàng)新互聯(lián)建站長(zhǎng)期為千余家客戶提供的網(wǎng)站建設(shè)服務(wù)，團(tuán)隊(duì)從業(yè)經(jīng)驗(yàn)10年，關(guān)注不同地域、不同群體，并針對(duì)不同對(duì)象提供差異化的產(chǎn)品和服務(wù)；打造開(kāi)放共贏平臺(tái)，與合作伙伴共同營(yíng)造健康的互聯(lián)網(wǎng)生態(tài)環(huán)境。為東昌企業(yè)提供專業(yè)的做網(wǎng)站、網(wǎng)站設(shè)計(jì)，東昌網(wǎng)站改版等技術(shù)服務(wù)。擁有十載豐富建站經(jīng)驗(yàn)和眾多成功案例,為您定制開(kāi)發(fā)。
ThreadPoolExecutor線程池，java提供開(kāi)發(fā)框架，管理線程的創(chuàng)建、銷毀、優(yōu)化、監(jiān)控等。
有4種不同的任務(wù)隊(duì)列:
1.ArrayBlockingQueue：基于數(shù)組結(jié)構(gòu)的任務(wù)隊(duì)列。此隊(duì)列按先進(jìn)先出的原則對(duì)任務(wù)進(jìn)行排序。
2.LinkedBlockingQueue：基于鏈表結(jié)構(gòu)的任務(wù)隊(duì)列。此隊(duì)列也是按先進(jìn)先出的原則對(duì)任務(wù)進(jìn)行排序。但性能比ArrayBlockingQueue高。
3.synchronousQueue：不存儲(chǔ)元素的任務(wù)隊(duì)列。每個(gè)插入操作必須等到另一個(gè)線程調(diào)用移除操作，否則插入操作一直處于阻塞狀態(tài)。
4.PriorityBlockingQueue：具有優(yōu)先級(jí)的任務(wù)隊(duì)列。此隊(duì)列中的元素必須能夠比較。
拒絕策略：
RejectedExecutionHandler（飽和策略 ）：當(dāng)線程池中的線程數(shù)大于maximumPoolSize時(shí)，線程池就不能在處理任何任務(wù)了，這時(shí)線程池會(huì)拋出異常。原因就是這個(gè)策略默認(rèn)情況下是AbortPolicy：表示無(wú)法處理新任務(wù)時(shí)拋出異常。
1.AbortPolicy：直接拋出異常。
2.CallerRunsPolicy：只用調(diào)用者所在線程來(lái)運(yùn)行任務(wù)。
3.DiscardOldestPolicy：丟棄隊(duì)列里最近的一個(gè)任務(wù)，并執(zhí)行當(dāng)前任務(wù)
4.DiscardPolicy：不處理，丟棄掉。
自定義：
ThreadPoolExecutor.AbortPolicy()
//拋出java.util.concurrent.RejectedExecutionException異常
ThreadPoolExecutor.CallerRunsPolicy()
//重試添加當(dāng)前的任務(wù)，他會(huì)自動(dòng)重復(fù)調(diào)用execute()方法
ThreadPoolExecutor.DiscardOldestPolicy()
//拋棄舊的任務(wù)
ThreadPoolExecutor.DiscardPolicy()
// 拋棄當(dāng)前的任務(wù)
重慶分公司
重慶分公司
private static class RecjectThreadHandler implements RejectedExecutionHandler
{
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
}
// 異常記錄
private void doLog(Runnable r, ThreadPoolExecutor executor)
{
System.out.println(r.toString()+"excutor failed."+executor.getCompletedTaskCount());
}
}創(chuàng)建線程工廠：
用來(lái)創(chuàng)建線程。
public class CheckThreadFactory implements ThreadFactory
{
private String threadGroupName;
private AtomicInteger count = new AtomicInteger(0);
public CheckThreadFactory(String threadGroupName) {
this.threadGroupName = threadGroupName;
}
@Override
public Thread newThread(Runnable r)
{
Thread thread = new Thread(r);
thread.setName(threadGroupName+"--"+count.addAndGet(1));
thread.setPriority(5);
thread.setDaemon(true);.// 設(shè)置為守護(hù)線程， 默認(rèn)為主線程
return thread;
}
}線程工具類：
/**
* @author Donald
* @create 2019-09-21 21:40
*/
public class CheckExcetPool
{
// 線程池核心線程數(shù)
private static int corePoolSize = Runtime.getRuntime().availableProcessors() * 5;
// 最大線程數(shù)
private static int maximumPoolSize = corePoolSize > 255 ? 255 : corePoolSize * 2;
// 線程池中除了核心線程，其他線程的最大存活時(shí)間
private static int keepAliveTime = 60;
// 時(shí)間單位
private static TimeUnit timeUnit = TimeUnit.SECONDS;
// 線程等待隊(duì)列
private static BlockingQueue queue = new LinkedBlockingQueue();
//private static BlockingQueue queue = new ArrayBlockingQueue(30);
// 創(chuàng)建線程的工廠
private static CheckThreadFactory checkThreadFactory = new CheckThreadFactory("checkGroup");
// 拒絕策略 當(dāng)提交任務(wù)數(shù)超過(guò)maxmumPoolSize+workQueue之和時(shí),
// * 即當(dāng)提交第41個(gè)任務(wù)時(shí)(前面線程都沒(méi)有執(zhí)行完,此測(cè)試方法中用sleep(100)),
// * 任務(wù)會(huì)交給RejectedExecutionHandler來(lái)處理
/*handler的拒絕策略：
有四種：第一種AbortPolicy:不執(zhí)行新任務(wù)，直接拋出異常，提示線程池已滿
第二種DisCardPolicy:不執(zhí)行新任務(wù)，也不拋出異常
第三種DisCardOldSetPolicy:將消息隊(duì)列中的第一個(gè)任務(wù)替換為當(dāng)前新進(jìn)來(lái)的任務(wù)執(zhí)行
第四種CallerRunsPolicy:直接調(diào)用execute來(lái)執(zhí)行當(dāng)前任務(wù)*/
private static ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
corePoolSize,
maximumPoolSize,
keepAliveTime,
timeUnit,
queue,
checkThreadFactory
);
public static void submit( Runnable runnable)
{
System.out.println(corePoolSize+"::"+queue.size());
threadPoolExecutor.submit(runnable);
}
public static Future submit(Callable callable)
{
return threadPoolExecutor.submit(callable);
}
public static void excutor( Runnable run, T result )
{
threadPoolExecutor.submit( run,result );
}
public static void excutor( Runnable run)
{
threadPoolExecutor.execute( run);
}
private static class RecjectThreadHandler implements RejectedExecutionHandler
{
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
}
// 異常記錄
private void doLog(Runnable r, ThreadPoolExecutor executor)
{
System.out.println(r.toString()+"excutor failed."+executor.getCompletedTaskCount());
}
}
} 線程服務(wù)類，實(shí)現(xiàn)runnable 接口：
/**
* @author Donald
* @create 2019-09-21 23:00
*/
public class ThreadService implements Runnable
{
private CountDownLatch countDownLatch;
private UserInterface userInterface;
public ThreadService(CountDownLatch countDownLatch, UserInterface userInterface) {
this.countDownLatch = countDownLatch;
this.userInterface = userInterface;
}
@Override
public void run()
{
try {
long start = System.currentTimeMillis();
userInterface.doSomething();
System.err.println(String.format("user time :%s",System.currentTimeMillis()-start));
Thread.sleep(1000);
}catch ( Exception e)
{
e.printStackTrace();
}finally {
countDownLatch.countDown();
}
}
}具體業(yè)務(wù)邏輯：
/**
* @author Donald
* @create 2019-09-21 22:51
*/
public interface UserInterface
{
void doSomething();
}業(yè)務(wù)類：
/**
* @author Donald
* @create 2019-09-21 22:51
*/
public class UserService implements UserInterface
{
private int number;
public UserService(int number) {
this.number = number;
}
@Override
public void doSomething() {
System.out.println(Thread.currentThread().getName()+"<<<<"+number);
}
}以上就是本文的全部?jī)?nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持創(chuàng)新互聯(lián)。
