一、说明
ThreadPoolExecutor
- 继承
Executor接口
- 它有多个构造方法来实现自定义创建线程池，以内部线程池的形式对外提供管理任务执行，线程调度，线程池管理等
- 关闭线程池调用
shutdown()、shutdownNow()、awaitTermination()方法
二、理解
shutdown()
- 只关闭了提交通道，停止接收新任务，已提交的任务会继续执行直到完成，此方法不会阻塞，当所有提交任务执行完毕，线程池被关闭
public void shutdown() { // 上锁确保只有一个线程执行此操作 final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { // 检查是否有权限关闭线程池以和中断线程 checkShutdownAccess(); // 将线程池状态设置为SHUTDOWN advanceRunState(SHUTDOWN); // 中断所有空闲线程 interruptIdleWorkers(); // 用于取消延时任务 onShutdown(); // hook for ScheduledThreadPoolExecutor } finally { mainLock.unlock(); } // 将线程池置为TERMINATED状态 tryTerminate(); }
shutdownNow()
- 立即停止线程池，停止接收新任务，中断所有正在执行的任务，停止对等待队列的处理，立刻返回未执行的任务列表
public List<Runnable> shutdownNow() { List<Runnable> tasks; // 上锁确保只有一个线程执行此操作 final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { // 检查是否有权限关闭线程池以和中断线程 checkShutdownAccess(); // 将线程池运行状态置为STOP advanceRunState(STOP); // 中断所有线程 interruptWorkers(); // 将未执行的任务移入列表中 tasks = drainQueue(); } finally { mainLock.unlock(); } // 将线程池置为TERMINATED状态 tryTerminate(); return tasks; }
awaitTermination()
- 在
shutdown()调用之后使用，阻塞当前线程，在这之后可以继续提交任务，设置等待超时时间，等待所有任务都执行完成，检查线程池是否终止，如果终止返回 true，否则返回 false，并解除阻塞
- 如果在超时之前所有任务执行完毕，表示线程池已经终止，返回true，否则返回false
- 如果在
shutdown()之前使用，线程池未终止，awaitTermination()锁在等待终止状态，造成死锁
public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException { // 将时间单位转化为纳秒 long nanos = unit.toNanos(timeout); // 上锁确保只有一个线程执行此操作 final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { // 如果线程池在超池之前终止，返回true while (!runStateAtLeast(ctl.get(), TERMINATED)) { // 超时但是线程池未终止，返回false if (nanos <= 0L) return false; // 实现阻塞 nanos = termination.awaitNanos(nanos); } return true; } finally { mainLock.unlock(); } }
线程池的生命周期
- 线程池的状态(runState)和工作线程数量(workerCount)共同保存在
AtomicInteger类型的控制变量ctl中
- ctl高三位保存运行状态(23=8>5)，低29位保存工作线程的数量(229-1)
// 初始运行状态为RUNNING，线程数为0 private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0)); // COUNT_BITS: 29 private static final int COUNT_BITS = Integer.SIZE - 3; // CAPACITY: 十进制: 536870911 二进制: 00011111111111111111111111111111 private static final int CAPACITY = (1 << COUNT_BITS) - 1; // runState is stored in the high-order bits // RUNNING: 十进制:-536870912 二进制:11100000000000000000000000000000 private static final int RUNNING = -1 << COUNT_BITS; // SHUTDOWN: 十进制:0 二进制:0 private static final int SHUTDOWN = 0 << COUNT_BITS; // STOP: 十进制:536870912 二进制:00100000000000000000000000000000 private static final int STOP = 1 << COUNT_BITS; // TIDYING: 十进制:1073741824 二进制:01000000000000000000000000000000 private static final int TIDYING = 2 << COUNT_BITS; // TERMINATED: 十进制:1610612736 二进制:01100000000000000000000000000000 private static final int TERMINATED = 3 << COUNT_BITS; // Packing and unpacking ctl 打包和解包ctl // 获取线程池当前状态，CAPACITY取反，高三位都是1，低29位都是0，和ctl进行与运算，获得runState变量 private static int runStateOf(int c) { return c & ~CAPACITY; } // CAPACITY高三位都是0，低29位都是0，和ctl进行与运算获得workerCount变量 private static int workerCountOf(int c) { return c & CAPACITY; } // 初始化ctl变量，runState和workerCount进行或运算后共同存储在一个变量中 private static int ctlOf(int rs, int wc) { return rs | wc; }
RUNNING接收新的任务，并且可执行队列里的任务
SHUTDOWN停止接收新任务，但可执行队列里的任务
STOP停止接收新任务，不执行队列里的任务，中断正在执行的任务
TIDYING所有任务都已终止，线程数为0，线程池变为TIDYING状态，会执行钩子函数terminated()，钩子方法是指使用一个抽象类实现接口，一个抽象类实现这个接口，需要的方法设置为abstract，其它设置为空方法
TERMINATED终止状态，表示线程池已终止，已经执行完terminated()钩子方法
判断当前线程池运行状态
// 判断线程池当前运行状态是否小于给定值 private static boolean runStateLessThan(int c, int s) { return c < s; } // 判断线程池当前运行状态是否大于等于给定值 private static boolean runStateAtLeast(int c, int s) { return c >= s; } // 判断线程池是否处于RUNNING状态 private static boolean isRunning(int c) { return c < SHUTDOWN; } // 判断线程池是否处于SHUTDOWN状态 public boolean isShutdown() { return ! isRunning(ctl.get()); } // 判断线程池是否处于TERMINATING状态 public boolean isTerminating() { int c = ctl.get(); return ! isRunning(c) && runStateLessThan(c, TERMINATED); } // 判断线程池是否处于TERMINATED状态 public boolean isTerminated() { return runStateAtLeast(ctl.get(), TERMINATED); }
运行状态转换关系
三、实现
1.shutdown()
创建一个ShutdownTest类，默认使用ThreadPoolExecutor.AbortPolicy拒绝策略，队列是ArrayBlockingQueue，设置核心线程数最大值为1，线程池线程数最大值为2，最大等待时间为5秒，等待队列值为2，提交8个任务，在第5个任务的时候执行 shutdown()
public class ShutdownTest { public static void main(String[] args) { // 1.创建线程池 ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(1, 2, 5, TimeUnit.SECONDS, new ArrayBlockingQueue<>(2), Executors.defaultThreadFactory()); // 2.创建线程任务 for (int i = 1; i <= 8; i++) { // 3.执行任务 System.out.println("执行第"+i+"个任务"); threadPoolExecutor.execute(new runnable("任务"+i)); // 4.获取等待队列 Iterator iterator = threadPoolExecutor.getQueue().iterator(); System.out.print("当前等待队列 "); while (iterator.hasNext()){ runnable thread = (runnable) iterator.next(); System.out.print(thread.name + "\t"); } System.out.print("\n"); System.out.println("--------"); // 5.关闭线程池 if (i == 4) { threadPoolExecutor.shutdown(); System.out.println("线程池已关闭"); } } } static class runnable implements Runnable{ // 设置任务名 String name; public runnable(String setName) { this.name = setName; } @Override public void run() { try { System.out.println("线程:"+Thread.currentThread().getName() +" 执行: "+name); } catch (Exception e) { e.printStackTrace(); } } } }
执行 shutdown()，此时停止接收新任务，已提交的任务会继续执行直到完成，此方法不会阻塞，抛出RejectedExecutionException
如果捕获RejectedExecutionException，可以看到任务被拒绝了
public static void main(String[] args) { // 1.创建线程池 ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(1, 2, 5, TimeUnit.SECONDS, new ArrayBlockingQueue<>(2), Executors.defaultThreadFactory()); // 2.创建线程任务 for (int i = 1; i <= 8; i++) { // 3.执行任务 System.out.println("执行第"+i+"个任务"); try { threadPoolExecutor.execute(new runnable("任务"+i)); // 4.获取等待队列 Iterator iterator = threadPoolExecutor.getQueue().iterator(); System.out.print("当前等待队列 "); while (iterator.hasNext()){ runnable thread = (runnable) iterator.next(); System.out.print(thread.name + "\t"); } System.out.print("\n"); System.out.println("--------"); } catch (RejectedExecutionException e) { // 5.捕获拒绝执行策略异常 System.out.println("拒绝执行第" + i + "个任务"); } // 6.关闭线程池 if (i == 4) { threadPoolExecutor.shutdown(); System.out.println("线程池已关闭"); } } }
2.shutdownNow()
立即停止线程池，停止接收新任务，中断所有正在执行的任务，停止对等待队列的处理
// 6.关闭线程池 if (i == 4) { threadPoolExecutor.shutdownNow(); System.out.println("线程池已关闭"); }
3.awaitTermination()
此方法阻塞，在shutdown()调用之后，停止接收新任务，但是awaitTermination()后可以继续提交，此方法是阻塞的，用来检测timeout时间后线程池是否终止，如果停止，则返回true并释放锁
public class ShutdownTest { public static void main(String[] args) throws InterruptedException { // 1.创建线程池 ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(1, 2, 5, TimeUnit.SECONDS, new ArrayBlockingQueue<>(2), Executors.defaultThreadFactory()); // 2.创建线程任务 for (int i = 1; i <= 8; i++) { // 3.执行任务 System.out.println("执行第"+i+"个任务"); try { threadPoolExecutor.execute(new runnable("任务"+i)); // 4.获取等待队列 Iterator iterator = threadPoolExecutor.getQueue().iterator(); System.out.print("当前等待队列 "); while (iterator.hasNext()){ runnable thread = (runnable) iterator.next(); System.out.print(thread.name + "\t"); } System.out.print("\n"); System.out.println("--------"); } catch (RejectedExecutionException e) { // 5.捕获拒绝执行策略异常 System.out.println("拒绝执行第" + i + "个任务"); } // 6.关闭线程池 if (i == 4) { threadPoolExecutor.shutdown(); while (!threadPoolExecutor.awaitTermination(1, TimeUnit.SECONDS)) { System.out.println("线程池未关闭"); } System.out.println("线程池已经关闭"); } } } static class runnable implements Runnable{ // 设置任务名 String name; public runnable(String setName) { this.name = setName; } @Override public void run() { try { System.out.println("线程:"+Thread.currentThread().getName() +" 执行: "+name); Thread.sleep(5000); } catch (Exception e) { e.printStackTrace(); } } } }
保证了线程池和其他线程的之间的执行顺序
如果 shutdown() 在 awaitTermination()后调用的话，awaitTermination()依然锁在等待终止状态，而 shutdown() 也无法得到锁去让线程池停止，这就形成了死锁
// 6.关闭线程池 if (i == 4) { while (!threadPoolExecutor.awaitTermination(1, TimeUnit.SECONDS)) { System.out.println("线程池未关闭"); } threadPoolExecutor.shutdown(); System.out.println("线程池已经关闭"); }
