写点什么

Java 多线程 关闭线程池 shutdown() 、shutdownNow()、awaitTermination()

作者:Yeats_Liao
  • 2022-10-18
    江西
  • 本文字数:5939 字

    阅读完需:约 1 分钟

一、说明

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 高三位保存运行状态(2^3^=8>5),低 29 位保存工作线程的数量(2^29^-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("线程池已经关闭");            }
复制代码



发布于: 2 小时前阅读数: 5
用户头像

Yeats_Liao

关注

Hello,World! 2022-10-02 加入

这里更多的是记录个人学习,如果有侵权内容请联系我! 个人邮箱是:yeats_liao@foxmail.com

评论

发布
暂无评论
Java多线程 关闭线程池 shutdown() 、shutdownNow()、awaitTermination()_后端_Yeats_Liao_InfoQ写作社区