java线程池ThreadPoolExecutor类使用详解

亦凉 2021-11-27 03:18 432阅读 0赞

本文转自：https://www.cnblogs.com/dafanjoy/p/9729358.html

在《阿里巴巴java开发手册》中指出了线程资源必须通过线程池提供，不允许在应用中自行显示的创建线程，这样一方面是线程的创建更加规范，可以合理控制开辟线程的数量；另一方面线程的细节管理交给线程池处理，优化了资源的开销。而线程池不允许使用Executors去创建，而要通过ThreadPoolExecutor方式，这一方面是由于jdk中Executor框架虽然提供了如newFixedThreadPool()、newSingleThreadExecutor()、newCachedThreadPool()等创建线程池的方法，但都有其局限性，不够灵活；另外由于前面几种方法内部也是通过ThreadPoolExecutor方式实现，使用ThreadPoolExecutor有助于大家明确线程池的运行规则，创建符合自己的业务场景需要的线程池，避免资源耗尽的风险。

下面我们就对ThreadPoolExecutor的使用方法进行一个详细的概述。

首先看下ThreadPoolExecutor的构造函数

  1. public ThreadPoolExecutor(int corePoolSize,
  2.                               int maximumPoolSize,
  3.                               long keepAliveTime,
  4.                               TimeUnit unit,
  5.                               BlockingQueue<Runnable> workQueue,
  6.                               ThreadFactory threadFactory,
  7.                               RejectedExecutionHandler handler) {
  8.         if (corePoolSize < 0 ||
  9.             maximumPoolSize <= 0 ||
  10.             maximumPoolSize < corePoolSize ||
  11.             keepAliveTime < 0)
  12.             throw new IllegalArgumentException();
  13.         if (workQueue == null || threadFactory == null || handler == null)
  14.             throw new NullPointerException();
  15.         this.acc = System.getSecurityManager() == null ?
  16.                 null :
  17.                 AccessController.getContext();
  18.         this.corePoolSize = corePoolSize;
  19.         this.maximumPoolSize = maximumPoolSize;
  20.         this.workQueue = workQueue;
  21.         this.keepAliveTime = unit.toNanos(keepAliveTime);
  22.         this.threadFactory = threadFactory;
  23.         this.handler = handler;
  24.     }

构造函数的参数含义如下：

corePoolSize:指定了线程池中的线程数量，它的数量决定了添加的任务是开辟新的线程去执行，还是放到workQueue任务队列中去；

maximumPoolSize:指定了线程池中的最大线程数量，这个参数会根据你使用的workQueue任务队列的类型，决定线程池会开辟的最大线程数量；

keepAliveTime:当线程池中空闲线程数量超过corePoolSize时，多余的线程会在多长时间内被销毁；

unit:keepAliveTime的单位

workQueue:任务队列，被添加到线程池中，但尚未被执行的任务；它一般分为直接提交队列、有界任务队列、无界任务队列、优先任务队列几种；

threadFactory:线程工厂，用于创建线程，一般用默认即可；

handler:拒绝策略；当任务太多来不及处理时，如何拒绝任务；

接下来我们对其中比较重要参数做进一步的了解：

一、workQueue任务队列

上面我们已经介绍过了，它一般分为直接提交队列、有界任务队列、无界任务队列、优先任务队列；

1、直接提交队列：设置为SynchronousQueue队列，SynchronousQueue是一个特殊的BlockingQueue，它没有容量，每执行一个插入操作就会阻塞，需要再执行一个删除操作才会被唤醒，反之每一个删除操作也都要等待对应的插入操作。

  1. public class ThreadPool {
  2.     private static ExecutorService pool;
  3.     public static void main( String[] args )
  4.     {
  5.         //maximumPoolSize设置为2 ，拒绝策略为AbortPolic策略，直接抛出异常
  6.         pool = new ThreadPoolExecutor(1, 2, 1000, 
  7.             TimeUnit.MILLISECONDS, new SynchronousQueue<Runnable>(),
  8.             Executors.defaultThreadFactory(),new ThreadPoolExecutor.AbortPolicy());
  9.         for(int i=0;i<3;i++) {
  10.             pool.execute(new ThreadTask());
  11.         }   
  12.     }
  13. }
  15. public class ThreadTask implements Runnable{
  17.     public ThreadTask() {
  19.     }
  21.     public void run() {
  22.         System.out.println(Thread.currentThread().getName());
  23.     }
  24. }

输出结果为

  1. pool-1-thread-1
  2. pool-1-thread-2
  3. Exception in thread "main" java.util.concurrent.RejectedExecutionException: 
  4. Task com.hhxx.test.ThreadTask@55f96302 rejected from java.util.concurrent.ThreadPoolExecutor@3d4eac69
  5. [Running, pool size = 2, active threads = 0, queued tasks = 0, completed tasks = 2]
  6.     at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(Unknown Source)
  7.     at java.util.concurrent.ThreadPoolExecutor.reject(Unknown Source)
  8.     at java.util.concurrent.ThreadPoolExecutor.execute(Unknown Source)
  9.     at com.hhxx.test.ThreadPool.main(ThreadPool.java:17)

可以看到，当任务队列为SynchronousQueue，创建的线程数大于maximumPoolSize时，直接执行了拒绝策略抛出异常。

使用SynchronousQueue队列，提交的任务不会被保存，总是会马上提交执行。如果用于执行任务的线程数量小于maximumPoolSize，则尝试创建新的进程，如果达到maximumPoolSize设置的最大值，则根据你设置的handler执行拒绝策略。因此这种方式你提交的任务不会被缓存起来，而是会被马上执行，在这种情况下，你需要对你程序的并发量有个准确的评估，才能设置合适的maximumPoolSize数量，否则很容易就会执行拒绝策略；

2、有界的任务队列：有界的任务队列可以使用ArrayBlockingQueue实现，如下所示

  1. pool = new ThreadPoolExecutor(3, 6, 1000, TimeUnit.MILLISECONDS, 
  2.             new ArrayBlockingQueue<Runnable>(5),Executors.defaultThreadFactory(),
  3.             new ThreadPoolExecutor.AbortPolicy());

向线程池中放入10个线程，输出结果：

  1. pool-1-thread-2
  2. pool-1-thread-2
  3. pool-1-thread-2
  4. pool-1-thread-2
  5. pool-1-thread-2
  6. pool-1-thread-2
  7. pool-1-thread-4
  8. pool-1-thread-1
  9. pool-1-thread-3
  10. pool-1-thread-5
  12. pool = new ThreadPoolExecutor(3, 6, 1000, TimeUnit.MILLISECONDS,
  13.                     new ArrayBlockingQueue<Runnable>(7), Executors.defaultThreadFactory(),
  14.                     new ThreadPoolExecutor.AbortPolicy());

向线程池中放入10个线程，输出结果为：（线程数维持corePoolSize）

  1. pool-1-thread-3
  2. pool-1-thread-3
  3. pool-1-thread-3
  4. pool-1-thread-3
  5. pool-1-thread-3
  6. pool-1-thread-3
  7. pool-1-thread-3
  8. pool-1-thread-3
  9. pool-1-thread-2
  10. pool-1-thread-1

向线程池中放入30个线程，输出结果为：

  1. pool-1-thread-2
  2. pool-1-thread-2
  3. pool-1-thread-2
  4. pool-1-thread-2
  5. pool-1-thread-2
  6. pool-1-thread-2
  7. pool-1-thread-2
  8. pool-1-thread-2
  9. pool-1-thread-1
  10. Exception in thread "main" pool-1-thread-5
  11. java.util.concurrent.RejectedExecutionException: Task 
  12. com.example.demo.threadpool.ThreadTask@776ec8df rejected from 
  13. java.util.concurrent.ThreadPoolExecutor@4eec7777
  14. [Running, pool size = 6, active threads = 3, queued tasks = 0, completed tasks = 10]
  15.     at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.
  16.             rejectedExecution(ThreadPoolExecutor.java:2047)
  17.         at java.util.concurrent.ThreadPoolExecutor.
  18.             reject(ThreadPoolExecutor.java:823)
  19.     at java.util.concurrent.ThreadPoolExecutor.
  20.             execute(ThreadPoolExecutor.java:1369)
  21.     at com.example.demo.threadpool.ThreadPoolTest.main(ThreadPoolTest.java:14)
  22. pool-1-thread-6
  23. pool-1-thread-3
  24. pool-1-thread-4

使用ArrayBlockingQueue有界任务队列，若有新的任务需要执行时，线程池会创建新的线程，直到创建的线程数量达到corePoolSize时，则会将新的任务加入到等待队列中。若等待队列已满，即超过ArrayBlockingQueue初始化的容量，则继续创建线程，直到线程数量达到maximumPoolSize设置的最大线程数量，若大于maximumPoolSize，则执行拒绝策略。在这种情况下，线程数量的上限与有界任务队列的状态有直接关系，如果有界队列初始容量较大或者没有达到超负荷的状态，线程数将一直维持在corePoolSize以下，反之当任务队列已满时，则会以maximumPoolSize为最大线程数上限。

3、无界的任务队列：有界任务队列可以使用LinkedBlockingQueue实现，如下所示

  1. pool = new ThreadPoolExecutor(1, 2, 1000, TimeUnit.MILLISECONDS, 
  2.         new LinkedBlockingQueue<Runnable>(),Executors.defaultThreadFactory(),
  3.         new ThreadPoolExecutor.AbortPolicy());

使用无界任务队列，线程池的任务队列可以无限制的添加新的任务，而线程池创建的最大线程数量就是你corePoolSize设置的数量，也就是说在这种情况下maximumPoolSize这个参数是无效的，哪怕你的任务队列中缓存了很多未执行的任务，当线程池的线程数达到corePoolSize后，就不会再增加了；若后续有新的任务加入，则直接进入队列等待，当使用这种任务队列模式时，一定要注意你任务提交与处理之间的协调与控制，不然会出现队列中的任务由于无法及时处理导致一直增长，直到最后资源耗尽的问题。

4、优先任务队列：优先任务队列通过PriorityBlockingQueue实现，下面我们通过一个例子演示下

  1. public class ThreadPool {
  2.     private static ExecutorService pool;
  3.     public static void main( String[] args )
  4.     {
  5.         //优先任务队列
  6.         pool = new ThreadPoolExecutor(1, 2, 1000, TimeUnit.MILLISECONDS, 
  7.             new PriorityBlockingQueue<Runnable>(),Executors.defaultThreadFactory(),
  8.             new ThreadPoolExecutor.AbortPolicy());
  10.         for(int i=0;i<20;i++) {
  11.             pool.execute(new ThreadTask(i));
  12.         }    
  13.     }
  14. }
  16. public class ThreadTask implements Runnable,Comparable<ThreadTask>{
  18.     private int priority;
  20.     public int getPriority() {
  21.         return priority;
  22.     }
  24.     public void setPriority(int priority) {
  25.         this.priority = priority;
  26.     }
  28.     public ThreadTask() {
  30.     }
  32.     public ThreadTask(int priority) {
  33.         this.priority = priority;
  34.     }
  36.     //当前对象和其他对象做比较，当前优先级大就返回-1，优先级小就返回1,值越小优先级越高
  37.     public int compareTo(ThreadTask o) {
  38.          return  this.priority>o.priority?-1:1;
  39.     }
  41.     public void run() {
  42.         try {
  43.             //让线程阻塞，使后续任务进入缓存队列
  44.             Thread.sleep(1000);
  45.             System.out.println("priority:"+this.priority+",
  46.                 ThreadName:"+Thread.currentThread().getName());
  47.         } catch (InterruptedException e) {
  48.             // TODO Auto-generated catch block
  49.             e.printStackTrace();
  50.         }
  52.     }
  53. }

我们来看下执行的结果情况

  1. priority:0,ThreadName:pool-1-thread-1
  2. priority:9,ThreadName:pool-1-thread-1
  3. priority:8,ThreadName:pool-1-thread-1
  4. priority:7,ThreadName:pool-1-thread-1
  5. priority:6,ThreadName:pool-1-thread-1
  6. priority:5,ThreadName:pool-1-thread-1
  7. priority:4,ThreadName:pool-1-thread-1
  8. priority:3,ThreadName:pool-1-thread-1
  9. priority:2,ThreadName:pool-1-thread-1
  10. priority:1,ThreadName:pool-1-thread-1

大家可以看到除了第一个任务直接创建线程执行外，其他的任务都被放入了优先任务队列，按优先级进行了重新排列执行，且线程池的线程数一直为corePoolSize，也就是只有一个。

通过运行的代码我们可以看出PriorityBlockingQueue它其实是一个特殊的无界队列，它其中无论添加了多少个任务，线程池创建的线程数也不会超过corePoolSize的数量，只不过其他队列一般是按照先进先出的规则处理任务，而PriorityBlockingQueue队列可以自定义规则根据任务的优先级顺序先后执行。

二、拒绝策略

一般我们创建线程池时，为防止资源被耗尽，任务队列都会选择创建有界任务队列，但种模式下如果出现任务队列已满且线程池创建的线程数达到你设置的最大线程数时，这时就需要你指定ThreadPoolExecutor的RejectedExecutionHandler参数即合理的拒绝策略，来处理线程池”超载”的情况。ThreadPoolExecutor自带的拒绝策略如下：

1、AbortPolicy策略：该策略会直接抛出异常，阻止系统正常工作；

2、CallerRunsPolicy策略：如果线程池的线程数量达到上限，该策略会把任务队列中的任务放在调用者线程当中运行；

3、DiscardOledestPolicy策略：该策略会丢弃任务队列中最老的一个任务，也就是当前任务队列中最先被添加进去的，马上要被执行的那个任务，并尝试再次提交；

4、DiscardPolicy策略：该策略会默默丢弃无法处理的任务，不予任何处理。当然使用此策略，业务场景中需允许任务的丢失；

以上内置的策略均实现了RejectedExecutionHandler接口，当然你也可以自己扩展RejectedExecutionHandler接口，定义自己的拒绝策略，我们看下示例代码：

  1. public class ThreadPool {
  2.     private static ExecutorService pool;
  3.     public static void main( String[] args )
  4.     {
  5.         //自定义拒绝策略
  6.         pool = new ThreadPoolExecutor(1, 2, 1000, TimeUnit.MILLISECONDS,
  7.                  new ArrayBlockingQueue<Runnable>(5),
  8.                 Executors.defaultThreadFactory(), 
  9.                 new RejectedExecutionHandler() {
  10.                     public void rejectedExecution(Runnable r, 
  11.                         ThreadPoolExecutor  executor) {
  12.                         System.out.println(r.toString()+"执行了拒绝策略");
  14.                     }
  15.                 });
  17.         for(int i=0;i<10;i++) {
  18.             pool.execute(new ThreadTask());
  19.         }    
  20.     }
  21. }
  23. public class ThreadTask implements Runnable{    
  24.     public void run() {
  25.         try {
  26.             //让线程阻塞，使后续任务进入缓存队列
  27.             Thread.sleep(1000);
  28.             System.out.println("ThreadName:"+Thread.currentThread().getName());
  29.         } catch (InterruptedException e) {
  30.             // TODO Auto-generated catch block
  31.             e.printStackTrace();
  32.         }
  34.     }
  35. }

输出结果：

  1. com.hhxx.test.ThreadTask@33909752执行了拒绝策略
  2. com.hhxx.test.ThreadTask@55f96302执行了拒绝策略
  3. com.hhxx.test.ThreadTask@3d4eac69执行了拒绝策略
  4. ThreadName:pool-1-thread-2
  5. ThreadName:pool-1-thread-1
  6. ThreadName:pool-1-thread-1
  7. ThreadName:pool-1-thread-2
  8. ThreadName:pool-1-thread-1
  9. ThreadName:pool-1-thread-2
  10. ThreadName:pool-1-thread-1

可以看到由于任务加了休眠阻塞，执行需要花费一定时间，导致会有一定的任务被丢弃，从而执行自定义的拒绝策略；

三、ThreadFactory自定义线程创建

线程池中线程就是通过ThreadPoolExecutor中的ThreadFactory，线程工厂创建的。那么通过自定义ThreadFactory，可以按需要对线程池中创建的线程进行一些特殊的设置，如命名、优先级等，下面代码我们通过ThreadFactory对线程池中创建的线程进行记录与命名

  1. public class ThreadPool {
  2.     private static ExecutorService pool;
  3.     public static void main( String[] args )
  4.     {
  5.         //自定义线程工厂
  6.         pool = new ThreadPoolExecutor(2, 4, 1000, TimeUnit.MILLISECONDS,
  7.                  new ArrayBlockingQueue<Runnable>(5),
  8.                  new ThreadFactory() {
  9.                     public Thread newThread(Runnable r) {
  10.                         System.out.println("线程"+r.hashCode()+"创建");
  11.                         //线程命名
  12.                         Thread th = new Thread(r,"threadPool"+r.hashCode());
  13.                         return th;
  14.                     }
  15.                 }, new ThreadPoolExecutor.CallerRunsPolicy());
  17.         for(int i=0;i<10;i++) {
  18.             pool.execute(new ThreadTask());
  19.         }    
  20.     }
  21. }
  23. public class ThreadTask implements Runnable{    
  24.     public void run() {
  25.         //输出执行线程的名称
  26.         System.out.println("ThreadName:"+Thread.currentThread().getName());
  27.     }
  28. }

输出结果：

  1. 线程118352462创建
  2. 线程1550089733创建
  3. 线程865113938创建
  4. ThreadName:threadPool1550089733
  5. ThreadName:threadPool118352462
  6. 线程1442407170创建
  7. ThreadName:threadPool1550089733
  8. ThreadName:threadPool1550089733
  9. ThreadName:threadPool1550089733
  10. ThreadName:threadPool865113938
  11. ThreadName:threadPool865113938
  12. ThreadName:threadPool118352462
  13. ThreadName:threadPool1550089733
  14. ThreadName:threadPool1442407170

可以看到线程池中，每个线程的创建我们都进行了记录输出与命名。

四、ThreadPoolExecutor扩展

ThreadPoolExecutor扩展主要是围绕beforeExecute()、afterExecute()和terminated()三个接口实现的，

1、beforeExecute：线程池中任务运行前执行

2、afterExecute：线程池中任务运行完毕后执行

3、terminated：线程池退出后执行

通过这三个接口我们可以监控每个任务的开始和结束时间，或者其他一些功能。下面我们可以通过代码实现一下

  1. public class ThreadPool {
  2.     private static ExecutorService pool;
  3.     public static void main( String[] args ) throws InterruptedException
  4.     {
  5.         //实现自定义接口
  6.         pool = new ThreadPoolExecutor(2, 4, 1000, TimeUnit.MILLISECONDS, 
  7.                 new ArrayBlockingQueue<Runnable>(5),
  8.                 new ThreadFactory() {
  9.                     public Thread newThread(Runnable r) {
  10.                         System.out.println("线程"+r.hashCode()+"创建");
  11.                         //线程命名
  12.                         Thread th = new Thread(r,"threadPool"+r.hashCode());
  13.                         return th;
  14.                     }
  15.                 }, new ThreadPoolExecutor.CallerRunsPolicy()) {
  17.             protected void beforeExecute(Thread t,Runnable r) {
  18.                 System.out.println("准备执行："+ ((ThreadTask)r).getTaskName());
  19.             }
  21.             protected void afterExecute(Runnable r,Throwable t) {
  22.                 System.out.println("执行完毕："+((ThreadTask)r).getTaskName());
  23.             }
  25.             protected void terminated() {
  26.                 System.out.println("线程池退出");
  27.             }
  28.         };
  30.         for(int i=0;i<10;i++) {
  31.             pool.execute(new ThreadTask("Task"+i));
  32.         }    
  33.         pool.shutdown();
  34.     }
  35. }
  37. public class ThreadTask implements Runnable{    
  38.     private String taskName;
  39.     public String getTaskName() {
  40.         return taskName;
  41.     }
  42.     public void setTaskName(String taskName) {
  43.         this.taskName = taskName;
  44.     }
  45.     public ThreadTask(String name) {
  46.         this.setTaskName(name);
  47.     }
  48.     public void run() {
  49.         //输出执行线程的名称
  50.         System.out.println("TaskName"+this.getTaskName()+
  51.             "---ThreadName:"+Thread.currentThread().getName());
  52.     }
  53. }

输出结果：

  1. 线程118352462创建
  2. 线程1550089733创建
  3. 准备执行：Task0
  4. 准备执行：Task1
  5. TaskNameTask0---ThreadName:threadPool118352462
  6. 线程865113938创建
  7. 执行完毕：Task0
  8. TaskNameTask1---ThreadName:threadPool1550089733
  9. 执行完毕：Task1
  10. 准备执行：Task3
  11. TaskNameTask3---ThreadName:threadPool1550089733
  12. 执行完毕：Task3
  13. 准备执行：Task2
  14. 准备执行：Task4
  15. TaskNameTask4---ThreadName:threadPool1550089733
  16. 执行完毕：Task4
  17. 准备执行：Task5
  18. TaskNameTask5---ThreadName:threadPool1550089733
  19. 执行完毕：Task5
  20. 准备执行：Task6
  21. TaskNameTask6---ThreadName:threadPool1550089733
  22. 执行完毕：Task6
  23. 准备执行：Task8
  24. TaskNameTask8---ThreadName:threadPool1550089733
  25. 执行完毕：Task8
  26. 准备执行：Task9
  27. TaskNameTask9---ThreadName:threadPool1550089733
  28. 准备执行：Task7
  29. 执行完毕：Task9
  30. TaskNameTask2---ThreadName:threadPool118352462
  31. TaskNameTask7---ThreadName:threadPool865113938
  32. 执行完毕：Task7
  33. 执行完毕：Task2
  34. 线程池退出

可以看到通过对beforeExecute()、afterExecute()和terminated()的实现，我们对线程池中线程的运行状态进行了监控，在其执行前后输出了相关打印信息。另外使用shutdown方法可以比较安全的关闭线程池， 当线程池调用该方法后，线程池中不再接受后续添加的任务。但是，此时线程池不会立刻退出，直到添加到线程池中的任务都已经处理完成，才会退出。

五、线程池线程数量

线程吃线程数量的设置没有一个明确的指标，根据实际情况，只要不是设置的偏大和偏小都问题不大，结合下面这个公式即可

  1. /**
  2.              * Nthreads=CPU数量
  3.              * Ucpu=目标CPU的使用率，0<=Ucpu<=1
  4.              * W/C=任务等待时间与任务计算时间的比率
  5.              */
  6.             Nthreads = Ncpu*Ucpu*(1+W/C)

发表评论

表情：
评论列表 （有 0 条评论，432人围观）

还没有评论，来说两句吧...

相关阅读

    相关 Java线程池ThreadPoolExecutor

    为什么使用线程池 如果每次使用线程都创建，每次创建和销毁的开销会很大，线程池主要用来解决线程生命周期开销问题和资源不足问题，也消除了线程创建所带来的延迟。 线

    你的名字你的名字/ 0 赞/ 329 阅读