一文搞懂ThreadPoolExecutor

迈不过友情╰ 2024-03-26 02:22 189阅读 0赞

前言

线程池是Java中使用较多的并发框架，合理使用线程池，可以：降低资源消耗，提高响应速度，提高线程的可管理性。

本篇文章将从线程池简单原理，线程池的创建，线程池执行任务和关闭线程池进行使用学习。

正文

一. 线程池的简单原理

当一个任务提交到线程池ThreadPoolExecutor时，该任务的执行如下图所示。

format_png

  1. 如果当前运行的线程数小于corePoolSzie（核心线程数），则创建新线程来执行任务（需要获取全局锁）；
  2. 如果当前运行的线程数等于或大于corePoolSzie，则将任务加入BlockingQueue（任务阻塞队列）；
  3. 如果BlockingQueue已满，则创建新的线程来执行任务（需要获取全局锁）；
  4. 如果创建新线程会使当前线程数大于maximumPoolSize（最大线程数），则拒绝任务并调用RejectedExecutionHandler的rejectedExecution() 方法。

由于ThreadPoolExecutor存储工作线程使用的集合是HashSet，因此执行上述步骤1和步骤3时需要获取全局锁来保证线程安全，而获取全局锁会导致线程池性能瓶颈，因此通常情况下，线程池完成预热后（当前线程数大于等于corePoolSize），线程池的execute() 方法都是执行步骤2。

二. 线程池的创建

通过ThreadPoolExecutor能够创建一个线程池，ThreadPoolExecutor的构造函数签名如下。

  1. public ThreadPoolExecutor(int corePoolSize,
  2.                           int maximumPoolSize,
  3.                           long keepAliveTime,
  4.                           TimeUnit unit,
  5.                           BlockingQueue<Runnable> workQueue)
  7. public ThreadPoolExecutor(int corePoolSize,
  8.                           int maximumPoolSize,
  9.                           long keepAliveTime,
  10.                           TimeUnit unit,
  11.                           BlockingQueue<Runnable> workQueue,
  12.                           ThreadFactory threadFactory)
  14. public ThreadPoolExecutor(int corePoolSize,
  15.                           int maximumPoolSize,
  16.                           long keepAliveTime,
  17.                           TimeUnit unit,
  18.                           BlockingQueue<Runnable> workQueue,
  19.                           RejectedExecutionHandler handler)
  21. public ThreadPoolExecutor(int corePoolSize,
  22.                           int maximumPoolSize,
  23.                           long keepAliveTime,
  24.                           TimeUnit unit,
  25.                           BlockingQueue<Runnable> workQueue,
  26.                           ThreadFactory threadFactory,
  27.                           RejectedExecutionHandler handler)

通过ThreadPoolExecutor创建线程池时，需要指定线程池的核心线程数，最大线程数，线程保活时间，线程保活时间单位和任务阻塞队列，并按需指定线程工厂和饱和拒绝策略，如果不指定线程工厂和饱和拒绝策略，则ThreadPoolExecutor会使用默认的线程工厂和饱和拒绝策略。下面分别介绍这些参数的含义。






































参数 含义
corePoolSize 核心线程数，即线程池的基本大小。当一个任务被提交到线程池时，如果线程池的线程数小于corePoolSize，那么无论其余线程是否空闲，也需创建一个新线程来执行任务。
maximumPoolSize 最大线程数。当线程池中线程数大于等于corePoolSize时，新提交的任务会加入任务阻塞队列，但是如果任务阻塞队列已满且线程数小于maximumPoolSize，此时会继续创建新的线程来执行任务。该参数规定了线程池允许创建的最大线程数
keepAliveTime 线程保活时间。当线程池的线程数大于核心线程数时，多余的空闲线程会最大存活keepAliveTime的时间，如果超过这个时间且空闲线程还没有获取到任务来执行，则该空闲线程会被回收掉。
unit 线程保活时间单位。通过TimeUnit指定线程保活时间的时间单位，可选单位有DAYS（天），HOURS（时），MINUTES（分），SECONDS（秒），MILLISECONDS（毫秒），MICROSECONDS（微秒）和NANOSECONDS（纳秒），但无论指定什么时间单位，ThreadPoolExecutor统一会将其转换为NANOSECONDS。
workQueue 任务阻塞队列。线程池的线程数大于等于corePoolSize时，新提交的任务会添加到workQueue中，所有线程执行完上一个任务后，会循环从workQueue中获取任务来执行。
threadFactory 创建线程的工厂。可以通过线程工厂给每个创建出来的线程设置更有意义的名字。
handler 饱和拒绝策略。如果任务阻塞队列已满且线程池中的线程数等于maximumPoolSize，说明线程池此时处于饱和状态，应该执行一种拒绝策略来处理新提交的任务。

三. 线程池执行任务

  1. 执行无返回值任务

通过ThreadPoolExecutor的execute() 方法，能执行Runnable任务，示例如下。

  1. public class ThreadPoolExecutorTest {
  3.     @Test
  4.     public void ThreadPoolExecutor执行简单无返回值任务() throws Exception {
  5.         // 创建一个线程池
  6.         ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2, 4,
  7.                 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(300));
  9.         // 创建两个任务
  10.         Runnable firstRunnable = new Runnable() {
  11.             @Override
  12.             public void run() {
  13.                 System.out.println("第一个任务执行");
  14.             }
  15.         };
  16.         Runnable secondRunnable = new Runnable() {
  17.             @Override
  18.             public void run() {
  19.                 System.out.println("第二个任务执行");
  20.             }
  21.         };
  23.         // 让线程池执行任务
  24.         threadPoolExecutor.execute(firstRunnable);
  25.         threadPoolExecutor.execute(secondRunnable);
  27.         // 让主线程睡眠1秒，等待线程池中的任务被执行完毕
  28.         Thread.sleep(1000);
  29.     }
  31. }

运行测试程序，结果如下。

format_png 1

  1. 执行有返回值任务

通过ThreadPoolExecutor的submit() 方法，能够执行Callable任务，通过submit() 方法返回的RunnableFuture能够拿到异步执行的结果。示例如下。

  1. public class ThreadPoolExecutorTest {
  3.     @Test
  4.     public void ThreadPoolExecutor执行简单有返回值任务() throws Exception {
  5.         // 创建一个线程池
  6.         ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2, 4,
  7.                 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(300));
  9.         // 创建两个任务，任务执行完有返回值
  10.         Callable<String> firstCallable = new Callable<String>() {
  11.             @Override
  12.             public String call() throws Exception {
  13.                 return "第一个任务返回值";
  14.             }
  15.         };
  16.         Callable<String> secondCallable = new Callable<String>() {
  17.             @Override
  18.             public String call() throws Exception {
  19.                 return "第二个任务返回值";
  20.             }
  21.         };
  23.         // 让线程池执行任务
  24.         Future<String> firstFuture = threadPoolExecutor.submit(firstCallable);
  25.         Future<String> secondFuture = threadPoolExecutor.submit(secondCallable);
  27.         // 获取执行结果，拿不到结果会阻塞在get()方法上
  28.         System.out.println(firstFuture.get());
  29.         System.out.println(secondFuture.get());
  30.     }
  32. }

运行测试程序，结果如下。

format_png 2

  1. 执行有返回值任务时抛出错误

如果ThreadPoolExecutor在执行Callable任务时，在Callable任务中抛出了异常并且没有捕获，那么这个异常是可以通过Future的get() 方法感知到的。示例如下。

  1. public class ThreadPoolExecutorTest {
  3.     @Test
  4.     public void ThreadPoolExecutor执行简单有返回值任务时抛出错误() {
  5.         // 创建一个线程池
  6.         ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2, 4,
  7.                 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(300));
  9.         // 创建一个任务，任务有返回值，但是执行过程中抛出异常
  10.         Callable<String> exceptionCallable = new Callable<String>() {
  11.             @Override
  12.             public String call() throws Exception {
  13.                 throw new RuntimeException("发生了异常");
  14.             }
  15.         };
  17.         // 让线程池执行任务
  18.         Future<String> exceptionFuture = threadPoolExecutor.submit(exceptionCallable);
  20.         try {
  21.             System.out.println(exceptionFuture.get());
  22.         } catch (Exception e) {
  23.             System.out.println(e.getMessage());
  24.         }
  25.     }
  27. }

运行测试程序，结果如下。

format_png 3

  1. ThreadPoolExecutor通过submit方式执行Runnable

ThreadPoolExecutor可以通过submit() 方法来运行Runnable任务，并且还可以异步获取执行结果。示例如下。

  1. public class ThreadPoolExecutorTest {
  3.     @Test
  4.     public void ThreadPoolExecutor通过submit的方式来提交并执行Runnable() throws Exception {
  5.         // 创建一个线程池
  6.         ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2, 4,
  7.                 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(300));
  9.         // 创建结果对象
  10.         MyResult myResult = new MyResult();
  11.         // 创建Runnable对象
  12.         Runnable runnable = new Runnable() {
  13.             @Override
  14.             public void run() {
  15.                 myResult.setResult("任务执行了");
  16.             }
  17.         };
  19.         // 通过ThreadPoolExecutor的submit()方法提交Runnable
  20.         Future<MyResult> resultFuture = threadPoolExecutor.submit(runnable, myResult);
  22.         // 获取执行结果
  23.         MyResult finalResult = resultFuture.get();
  24.         // myResult和finalResult的地址实际相同
  25.         Assert.assertEquals(myResult, finalResult);
  26.         // 打印执行结果
  27.         System.out.println(resultFuture.get().getResult());
  28.     }
  30.     private static class MyResult {
  31.         String result;
  33.         public MyResult() {}
  35.         public MyResult(String result) {
  36.             this.result = result;
  37.         }
  39.         public String getResult() {
  40.             return result;
  41.         }
  43.         public void setResult(String result) {
  44.             this.result = result;
  45.         }
  46.     }
  48. }

运行测试程序，结果如下。

format_png 4

实际上ThreadPoolExecutor的submit() 方法无论是提交Runnable任务还是Callable任务，都是将任务封装成了RunnableFuture接口的子类FutureTask，然后调用ThreadPoolExecutor的execute() 方法来执行FutureTask。

四. 关闭线程池

关闭线程池可以通过ThreadPoolExecutor的shutdown() 方法，但是shutdown() 方法不会去中断正在执行任务的线程，所以如果线程池里有Worker正在执行一个永远不会结束的任务，那么shutdown() 方法是无法关闭线程池的。示例如下。

  1. public class ThreadPoolExecutorTest {
  3.     @Test
  4.     public void 通过shutdown关闭线程池() {
  5.         // 创建一个线程池
  6.         ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2, 4,
  7.                 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(300));
  9.         // 创建Runnable对象
  10.         Runnable runnable = new Runnable() {
  11.             @Override
  12.             public void run() {
  13.                 while (!Thread.currentThread().isInterrupted()) {
  14.                     LockSupport.parkNanos(1000 * 1000 * 1000);
  15.                 }
  16.                 System.out.println(Thread.currentThread().getName() + " 被中断");
  17.             }
  18.         };
  20.         // 让线程池执行任务
  21.         threadPoolExecutor.execute(runnable);
  22.         threadPoolExecutor.execute(runnable);
  24.         // 调用shutdown方法关闭线程池
  25.         threadPoolExecutor.shutdown();
  27.         // 等待3秒观察现象
  28.         LockSupport.parkNanos(1000 * 1000 * 1000 * 3L);
  29.     }
  31. }

运行测试程序，会发现在主线程中等待3秒后，也没有得到预期的打印结果。如果上述测试程序中使用shutdownNow，则是可以得到预期打印结果的，示例如下。

  1. public class ThreadPoolExecutorTest {
  3.     @Test
  4.     public void 通过shutdownNow关闭线程池() {
  5.         // 创建一个线程池
  6.         ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2, 4,
  7.                 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(300));
  9.         // 创建Runnable对象
  10.         Runnable runnable = new Runnable() {
  11.             @Override
  12.             public void run() {
  13.                 while (!Thread.currentThread().isInterrupted()) {
  14.                     LockSupport.parkNanos(1000 * 1000 * 1000);
  15.                 }
  16.                 System.out.println(Thread.currentThread().getName() + " 被中断");
  17.             }
  18.         };
  20.         // 让线程池执行任务
  21.         threadPoolExecutor.execute(runnable);
  22.         threadPoolExecutor.execute(runnable);
  24.         // 调用shutdown方法关闭线程池
  25.         threadPoolExecutor.shutdownNow();
  27.         // 等待3秒观察现象
  28.         LockSupport.parkNanos(1000 * 1000 * 1000 * 3L);
  29.     }
  31. }

运行测试程序，打印如下。

format_png 5

因为测试程序中的任务是响应中断的，而ThreadPoolExecutor的shutdownNow() 方法会中断所有Worker，所以执行shutdownNow() 方法后，正在运行的任务会响应中断并结束运行，最终线程池关闭。

假如线程池中运行着一个永远不会结束的任务，且这个任务不响应中断，那么无论是shutdown() 方法还是shutdownNow() 方法，都是无法关闭线程池的。

总结

ThreadPoolExecutor的使用总结如下。

  1. 通过ThreadPoolExecutor的execute() 方法能够执行Runnable任务；
  2. 通过ThreadPoolExecutor的submit() 方法能够执行Runnable任务和Callable任务，并且能够获取异步的执行结果；
  3. ThreadPoolExecutor的submit() 方法会返回一个Future对象（实际就是FutureTask），如果任务执行过程中发生了异常且未捕获，那么可以通过Future的get() 方法感知到异常；
  4. ThreadPoolExecutor的submit() 方法无论是提交Runnable任务还是Callable任务，都是将任务封装成了RunnableFuture接口的子类FutureTask，然后调用ThreadPoolExecutor的execute() 方法来执行FutureTask；
  5. 关闭线程池时，如果运行的任务可以在有限时间内运行完毕，那么可以使用shutdown() 方法来关闭线程池，这能够保证在关闭线程池时，正在运行的任务会顺利运行完毕；
  6. 关闭线程池时，如果运行的任务永远不会结束但是响应中断，那么可以使用shutdownNow() 方法来关闭线程池，这种方式不保证任务顺利运行完毕；
  7. 如果任务永远不会结束且不响应中断，那么无论是shutdown() 方法还是shutdownNow() 方法，都无法关闭线程池。

发表评论

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

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

相关阅读

    相关 一文搞懂static（一）

    一、聊一聊static与JVM Java 把内存分为栈内存和堆内存，其中栈内存用来存放一些基本类型的变量、数组和对象的引用，堆内存主要存放一些对象。在 JVM 加载一个类的

    怼烎@怼烎@/ 0 赞/ 195 阅读

    相关 一文搞懂ThreadPoolExecutor

    前言 线程池是Java中使用较多的并发框架，合理使用线程池，可以：降低资源消耗，提高响应速度，提高线程的可管理性。 本篇文章将从线程池简单原理，线程池的创建，线程池执行

    迈不过友情╰迈不过友情╰/ 0 赞/ 190 阅读

    相关 一文轻松搞懂 Docker

    什么是Docker？ Docker是一个开源的应用容器引擎，它基于go语言开发，并遵从Apache2.0开源协议。使用Docker可以让开发者封装他们的应用以及依赖包到一个可

    末蓝、末蓝、/ 0 赞/ 235 阅读

    相关 一文搞懂 Raft 算法

    一文搞懂Raft算法 正文 raft是工程上使用较为广泛的强一致性、去中心化、高可用的分布式协议。在这里强调了是在工程上，因为在学术理论界，最耀眼的还是大名鼎鼎的Pax

    不念不忘少年蓝@不念不忘少年蓝@/ 0 赞/ 429 阅读

    相关 一文搞懂DDD

    最近看了很多和DDD相关的内容，这篇文章对DDD做一个总结，希望可以通过这篇文章不但知道什么是DDD，而且还可以知道如何实施DDD 一、什么是DDD DDD（领域驱动设

    ゝ一世哀愁。ゝ一世哀愁。/ 0 赞/ 319 阅读