一、说明

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³=8>5)，低29位保存工作线程的数量(2²⁹-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("线程池已经关闭");
            }

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

一、说明

二、理解

三、实现

1.shutdown()

2.shutdownNow()

3.awaitTermination()

热门文章

最新文章

相关课程

相关电子书

相关实验场景

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

一、说明

二、理解

三、实现

1.shutdown()

2.shutdownNow()

3.awaitTermination()

热门文章

最新文章

相关课程

相关电子书

相关实验场景

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