ThreadPoolExecutor線程池的使用方法

更新時間：2019年09月25日 08:35:27 作者：灬奔跑的蝸牛灬

這篇文章主要為大家詳細介紹了ThreadPoolExecutor線程池的使用方法，具有一定的參考價值，感興趣的小伙伴們可以參考一下

ThreadPoolExecutor

ThreadPoolExecutor線程池，java提供開發(fā)框架，管理線程的創(chuàng)建、銷毀、優(yōu)化、監(jiān)控等。

有4種不同的任務隊列:

1.ArrayBlockingQueue：基于數(shù)組結構的任務隊列。此隊列按先進先出的原則對任務進行排序。

2.LinkedBlockingQueue：基于鏈表結構的任務隊列。此隊列也是按先進先出的原則對任務進行排序。但性能比ArrayBlockingQueue高。

3.synchronousQueue：不存儲元素的任務隊列。每個插入操作必須等到另一個線程調用移除操作，否則插入操作一直處于阻塞狀態(tài)。

4.PriorityBlockingQueue：具有優(yōu)先級的任務隊列。此隊列中的元素必須能夠比較。

拒絕策略：

RejectedExecutionHandler（飽和策略）：當線程池中的線程數(shù)大于maximumPoolSize時，線程池就不能在處理任何任務了，這時線程池會拋出異常。原因就是這個策略默認情況下是AbortPolicy：表示無法處理新任務時拋出異常。

1.AbortPolicy：直接拋出異常。

2.CallerRunsPolicy：只用調用者所在線程來運行任務。

3.DiscardOldestPolicy：丟棄隊列里最近的一個任務，并執(zhí)行當前任務

4.DiscardPolicy：不處理，丟棄掉。

自定義：

ThreadPoolExecutor.AbortPolicy()
//拋出java.util.concurrent.RejectedExecutionException異常
ThreadPoolExecutor.CallerRunsPolicy()
//重試添加當前的任務，他會自動重復調用execute()方法
ThreadPoolExecutor.DiscardOldestPolicy()
//拋棄舊的任務
ThreadPoolExecutor.DiscardPolicy()
// 拋棄當前的任務

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)建線程工廠：

用來創(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);.// 設置為守護線程， 默認為主線程
 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;
 // 線程池中除了核心線程，其他線程的最大存活時間
 private static int keepAliveTime = 60;
 // 時間單位
 private static TimeUnit timeUnit = TimeUnit.SECONDS;
 // 線程等待隊列
 private static BlockingQueue queue = new LinkedBlockingQueue();
 //private static BlockingQueue queue = new ArrayBlockingQueue<Runnable>(30);
 // 創(chuàng)建線程的工廠
 private static CheckThreadFactory checkThreadFactory = new CheckThreadFactory("checkGroup");
 // 拒絕策略 當提交任務數(shù)超過maxmumPoolSize+workQueue之和時,
 // * 即當提交第41個任務時(前面線程都沒有執(zhí)行完,此測試方法中用sleep(100)),
 // *   任務會交給RejectedExecutionHandler來處理
 /*handler的拒絕策略：
 有四種：第一種AbortPolicy:不執(zhí)行新任務，直接拋出異常，提示線程池已滿
 第二種DisCardPolicy:不執(zhí)行新任務，也不拋出異常
 第三種DisCardOldSetPolicy:將消息隊列中的第一個任務替換為當前新進來的任務執(zhí)行
 第四種CallerRunsPolicy:直接調用execute來執(zhí)行當前任務*/

 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 <T> Future submit(Callable<T> callable)
 {
 return threadPoolExecutor.submit(callable);
 }
 public static <T> 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());
 }
 }
}

線程服務類，實現(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è)務邏輯：

/**
 * @author Donald
 * @create 2019-09-21 22:51
 */
public interface UserInterface
{
 void doSomething();
}

業(yè)務類：

/**
 * @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);
 }

}

以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支持腳本之家。

您可能感興趣的文章: