Java并发计算判断线程池中的线程是否全部执行完毕

2023-09-21 104

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

简介： Java并发计算判断线程池中的线程是否全部执行完毕

1 java多线程的使用

Java多线程实现与应用： https://blog.csdn.net/ZGL_cyy/article/details/107756399

Java线程池ExecutorService：https://blog.csdn.net/ZGL_cyy/article/details/117843472

Java线程池七大参数详解和配置：https://blog.csdn.net/ZGL_cyy/article/details/118230264

2 shutdown和shutdownNow区别源码解析

1、这两个都是关闭线程池的方法，但是大家可能对其作用和原理不是很清楚，不知道线程池是否真的关闭了，或者又重新提交了任务会怎样？下面我先通过一些案列带大家看一下他们之间的一些区别

2.1 shutdown案例

    @Test
    public void shutdownTest(){
        ThreadPoolExecutor pool = new ThreadPoolExecutor(1, 2,
                1, TimeUnit.SECONDS, new LinkedBlockingQueue<>(2));
        Runnable runnable = () -> {
            try {
                Thread.sleep(1000);
                System.out.println("线程:" + Thread.currentThread().getName() + "正在执行");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        };
        pool.submit(runnable);
        pool.submit(runnable);
        pool.submit(runnable);
        pool.shutdown();
        System.out.println("main线程已经执行到这了");
        System.out.println("线程数："+pool.getActiveCount()+",队列里的任务："+pool.getQueue().size());
        try {
            TimeUnit.SECONDS.sleep(5);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

可以看出线程池虽然调用了shutdown方法，这个方法会将线程池的状态改为SHUTDOWN状态，不接受新的任务，但是会继续处理没有处理完的任务和阻塞队列中的任务，如果有新的任务过来会抛

RejectedExecutionException异常。

下面是我又提交了一个任务，然后不出所料的报RejectedExecutionException异常

2.2 shutdownNow案例

可能有人看方法名字就很快地回答出了，立刻关闭线程池，正在执行的任务和阻塞队列中没有执行的任务都不会执行了，那你就想错了，实际上并不是这样，下面看一个案例否定这种说法。

    @Test
    public void shutdownNowTest1(){
        ThreadPoolExecutor pool = new ThreadPoolExecutor(1, 2,
                1, TimeUnit.SECONDS, new LinkedBlockingQueue<>(2));
        Runnable runnable = () -> {
            Integer sum = 0;
            for(Integer i=0;i<10000;i++){
                sum += i;
            }
            System.out.println("sum:"+sum);
        };
        pool.submit(runnable);
        pool.submit(runnable);
        pool.shutdownNow();
        System.out.println("main已经走到这里了");
        try {
            TimeUnit.SECONDS.sleep(5);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

看到了吧这里main方法都打印了，说明已经已经执行完shutdownNow()方法了，但是为什么sum还能计算并且能输出呢，这就能说明正在执行的线程不是立即关闭。

其实shutdownNow()的方法底层循环遍历了工作线程的interrupt()方法，这个方法会Just to set the interrupt flag，然后线程遇到阻塞方法(sleep,join,wait等)就会抛出InterruptedException()异常，案例如下

    @Test
    public void shutdownNowTest1(){
        ThreadPoolExecutor pool = new ThreadPoolExecutor(1, 2,
                1, TimeUnit.SECONDS, new LinkedBlockingQueue<>(2));
        Runnable runnable = () -> {
            try {
                Thread.sleep(1000);
                System.out.println("111");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(222);
        };
        pool.submit(runnable);
        pool.submit(runnable);
        pool.shutdownNow();
        System.out.println("main已经走到这里了");
        try {
            TimeUnit.SECONDS.sleep(5);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

2.3 源码解读

下面来看一下他们的源码，有助于大家进一步记忆。

*shutdown源码*
    public void shutdown() {
        final ReentrantLock mainLock = this.mainLock;
        //加锁，同一时刻只能由一个线程执行线程池的shutdown()方法
        mainLock.lock();
        try {
          //检查调用方是否有权限关闭线程池和中断工作线程
            checkShutdownAccess();
            //将线程池的状态改为SHUTDOWN
            advanceRunState(SHUTDOWN);
            //中断所有空闲线程
            interruptIdleWorkers();
            //取消延时任务 ScheduledThreadPoolExecutor
            onShutdown(); // hook for ScheduledThreadPoolExecutor
        } finally {
          //释放锁
            mainLock.unlock();
        }
        //尝试将线程池置为TERMINATED
        tryTerminate();
    }
  //通过 CAS 自旋操作将线程池运行状态设置为目标值，如果已经大于等于目标值，则不作任何操作
  private void advanceRunState(int targetState) {
        for (;;) {
            int c = ctl.get();
            //如果当前运行状态 >= targetState，则直接break，不作处理
            //如果如果当前运行状态 < targetState,则通过CAS操作将运行状态设置为目标值，成功的话会返回true，失败则返回false
            if (runStateAtLeast(c, targetState) ||
                ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c))))
                break;
        }
    }
  private void interruptIdleWorkers() {
        interruptIdleWorkers(false);
    }
    //如果为true,则只中断一个空闲线程，如果为false,则中断所有空闲线程
    private void interruptIdleWorkers(boolean onlyOne) {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            for (Worker w : workers) {
                Thread t = w.thread;
                //线程没有中断，并且线程想以独占的方式获取锁，获取到返回true,表示线程处于空闲状态，没获取到返回false，表示线程处于工作状态
                //tryLock()底层调用了tryAcquire（），再底层就是AQS了
                if (!t.isInterrupted() && w.tryLock()) {
                    try {
                        t.interrupt();
                    } catch (SecurityException ignore) {
                    } finally {
                        w.unlock();
                    }
                }
                if (onlyOne)
                    break;
            }
        } finally {
            mainLock.unlock();
        }
    }
  //AQS源码
  protected boolean tryAcquire(int unused) {
    //先通过CAS尝试将同步状态(AQS的state属性)从0修改为1。若直接修改成功了，则将占用锁的线程设置为当前线程
        if (compareAndSetState(0, 1)) {
           //用来保存当前占用同步状态的线程。
             setExclusiveOwnerThread(Thread.currentThread());
             return true;
        }
        return false;
    }
  final void tryTerminate() {
        for (;;) {
            int c = ctl.get();
            if (isRunning(c) ||
                runStateAtLeast(c, TIDYING) ||
                (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
                return;
            //循环关闭正在工作的线程
            if (workerCountOf(c) != 0) { // Eligible to terminate
                interruptIdleWorkers(ONLY_ONE);
                return;
            }
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
              //将线程池状态置为TIDYING状态
                if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
                    try {
                        terminated();
                    } finally {
                        ctl.set(ctlOf(TERMINATED, 0));
                        termination.signalAll();
                    }
                    return;
                }
            } finally {
                mainLock.unlock();
            }
            // else retry on failed CAS
        }
    }
*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();
        }
        tryTerminate();
        return tasks;
    }
    private void interruptWorkers() {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
          //中断所有线程(interrupt flag = true)
            for (Worker w : workers)
                w.interruptIfStarted();
        } finally {
            mainLock.unlock();
        }
    }
    //中断线程，包括正在执行的线程（interrupt flag = true）
    void interruptIfStarted() {
            Thread t;
            if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
                try {
                    t.interrupt();
                } catch (SecurityException ignore) {
                }
            }
        }
    private List<Runnable> drainQueue() {
        BlockingQueue<Runnable> q = workQueue;
        ArrayList<Runnable> taskList = new ArrayList<Runnable>();
        //调用BlockingQueue的drainTo()方法转移元素
        q.drainTo(taskList);
        if (!q.isEmpty()) {
            for (Runnable r : q.toArray(new Runnable[0])) {
                if (q.remove(r))
                    taskList.add(r);
            }
        }
        return taskList;
    }

上面可以看出线程池调用shutdownNow()方法之后有三个关键步骤

会将线程池置于stop状态

然后会遍历正在执行的任务，调用正在执行线程的interrupt()方法

最后会将阻塞队列中未执行的任务放到task中返回。

通过源码可以看出shutdownNow()方法是有返回值的，返回的内容就是阻塞队列为执行的任务。而且shutdownNow()是不能直接停止当前工作线程（仅仅调用的interrupt()而不是stop())，若想调用shutdownNow()停止工作中的线程，只有判断判断线程的状态(flag=Thread.currentThread().isInterrupted())看线程是否关闭，或者调用能中断线程的方法（join,sleep,wait等）。但是不推荐关停正在执行的任务，因为线程的stop()方法都弃用了，官方解释说stop()会释放这个线程所持有的所有的锁对象这会导致对象处于不一致状态。

比如：可能会导致从一个账户向另一个账户转账的过程中被终止，钱已经转出，但没有输入目标账户。

2.4 shutdown和shutdownNow总结

shutdown():

1、同一时刻只能有一个线程能修改线程池的状态

2、检查调用方是否有权限关闭线程池和中断工作线程

3、通过CAS将线程池状态置为SHUTDOWN

4、通过AQS关闭所有空闲的线程

5、取消延时队列中的任务

6、尝试将线程池状态置为Terminated

shutdownNow():

1、同一时刻只能有一个线程能修改线程池的状态

2、检查调用方是否有权限关闭线程池和中断工作线程

3、通过CAS将线程池状态置为STOP

4、遍历所有正在执行的工作线程，调用其interrupt()方法，修改其中断状态标识为true

5、返回阻塞队列中未执行的任务给调用方

6、尝试将线程池状态置为Terminated

shutdown()和shutdownNow()方法都可以关闭线程池，但是工作中为了代码看起来比较优雅一般我们用shutdown()和awaitTermination()结合使用来关闭线程池，如果情况紧急，那么就用shutdownNow()来关闭线程池。

pool.shutdown();
pool.awaitTermination(Integer.MAX_VALUE, TimeUnit.SECONDS);
    @Test
    public void shutdownTest(){
        ThreadPoolExecutor pool = new ThreadPoolExecutor(1, 2,
                1, TimeUnit.SECONDS, new LinkedBlockingQueue<>(2));
        Runnable runnable = () -> {
            try {
                Thread.sleep(1000);
                System.out.println("线程:" + Thread.currentThread().getName() + "正在执行");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        };
        TimeInterval timer = DateUtil.timer();
        pool.submit(runnable);
        pool.submit(runnable);
        pool.submit(runnable);
        pool.shutdown();
        System.out.println("main线程已经执行到这了");
        System.out.println("线程数："+pool.getActiveCount()+",队列里的任务："+pool.getQueue().size());
        try {
            pool.awaitTermination(5,TimeUnit.SECONDS);
            System.out.println(timer.intervalRestart());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

3 判断线程池中的线程是否全部执行完毕

在使用多线程中我们会使用 java.util.concurrent.Executors的线程池，当多个子线程异步执行的时候,调用ExecutorService.shutdown方法，线程池不再接收任何新的任务，但此时线程池并不会立刻退出，直到添加到线程池中的任务都已经执行处理完成，才会退出,我们如何判断线程池中所有的子线程都已经执行完毕,然后继续执行后续操作。

3.1 方式一：调用ExecutorService 中的isTerminated方法

调用ExecutorService 中的isTerminated方法。

在调用shutdown方法后我们可以在一个死循环里面用isTerminated方法判断是否线程池中的所有线程已经执行完毕

3.2 方式二：调用ExecutorService 中的awaitTermination()方法,等待子线程结束

调用ExecutorService 中的awaitTermination()方法,等待子线程结束

3.3 方式三：闭锁CountDownLatch

闭锁CountDownLatch

闭锁状态包括一个计数器，该计数器被初始化为一个正数，表示需要等待的事件数量,递减计数,而await方法等待计数器达到零，即表示线程结束

Java并发计算判断线程池中的线程是否全部执行完毕

1 java多线程的使用

2 shutdown和shutdownNow区别源码解析

2.1 shutdown案例

2.2 shutdownNow案例

2.3 源码解读

2.4 shutdown和shutdownNow总结

3 判断线程池中的线程是否全部执行完毕

3.1 方式一：调用ExecutorService 中的isTerminated方法

3.2 方式二：调用ExecutorService 中的awaitTermination()方法,等待子线程结束

3.3 方式三：闭锁CountDownLatch

热门文章

最新文章

相关课程

相关电子书

相关实验场景

热门

活动广场

任务中心

开发者评测

高校计划

乘风者计划

训练营

阿里云MVP

话题

直播

下载

镜像站

技术资料

插件

Java并发计算判断线程池中的线程是否全部执行完毕

1 java多线程的使用

2 shutdown和shutdownNow区别源码解析

2.1 shutdown案例

2.2 shutdownNow案例

2.3 源码解读

2.4 shutdown和shutdownNow总结

3 判断线程池中的线程是否全部执行完毕

3.1 方式一：调用ExecutorService 中的isTerminated方法

3.2 方式二：调用ExecutorService 中的awaitTermination()方法,等待子线程结束

3.3 方式三：闭锁CountDownLatch

热门文章

最新文章

相关课程

相关电子书

相关实验场景