ThreadPoolExecutor线程池的使用方法

ThreadPoolExecutor

ThreadPoolExecutor线程池，java提供开发框架，管理线程的创建、销毁、优化、监控等。

有4种不同的任务队列:

1.ArrayBlockingQueue：基于数组结构的任务队列。此队列按先进先出的原则对任务进行排序。

2.LinkedBlockingQueue：基于链表结构的任务队列。此队列也是按先进先出的原则对任务进行排序。但性能比ArrayBlockingQueue高。

3.synchronousQueue：不存储元素的任务队列。每个插入操作必须等到另一个线程调用移除操作，否则插入操作一直处于阻塞状态。

4.PriorityBlockingQueue：具有优先级的任务队列。此队列中的元素必须能够比较。

拒绝策略：

RejectedExecutionHandler（饱和策略 ）：当线程池中的线程数大于maximumPoolSize时，线程池就不能在处理任何任务了，这时线程池会抛出异常。原因就是这个策略默认情况下是AbortPolicy：表示无法处理新任务时抛出异常。

1.AbortPolicy：直接抛出异常。

2.CallerRunsPolicy：只用调用者所在线程来运行任务。

3.DiscardOldestPolicy：丢弃队列里最近的一个任务，并执行当前任务

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());
    }
  }

创建线程工厂：

用来创建线程。

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
{
  // 线程池核心线程数
  private static int corePoolSize = Runtime.getRuntime().availableProcessors() * 5;
  // 最大线程数
  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);
  // 创建线程的工厂
  private static CheckThreadFactory checkThreadFactory = new CheckThreadFactory("checkGroup");
  // 拒绝策略 当提交任务数超过maxmumPoolSize+workQueue之和时,//   *    即当提交第41个任务时(前面线程都没有执行完,此测试方法中用sleep(100)),//   *         任务会交给RejectedExecutionHandler来处理
  /*handler的拒绝策略：
  有四种：第一种AbortPolicy:不执行新任务，直接抛出异常，提示线程池已满
  第二种DisCardPolicy:不执行新任务，也不抛出异常
  第三种DisCardOldSetPolicy:将消息队列中的第一个任务替换为当前新进来的任务执行
  第四种CallerRunsPolicy:直接调用execute来执行当前任务*/

  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)
    {
      System.out.println(r.toString()+"excutor failed."+executor.getCompletedTaskCount());
    }
  }
}

线程服务类，实现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();
    }
  }
}

具体业务逻辑：

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

业务类：

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

}

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持我们。

（编辑：李大同）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!