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

简介: 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();
        }
    }

f0c2d9d0f6094e10a995bacbef24da9d.png

看到了吧这里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方法等待计数器达到零,即表示线程结束

目录
相关文章
|
20小时前
|
缓存 Java 调度
Java并发编程:深入理解线程池
【4月更文挑战第30天】 在Java并发编程中,线程池是一种重要的工具,它可以帮助我们有效地管理线程,提高系统性能。本文将深入探讨Java线程池的工作原理,如何使用它,以及如何根据实际需求选择合适的线程池策略。
|
1天前
|
Java
Java并发编程:深入理解线程池
【4月更文挑战第30天】 本文将深入探讨Java中的线程池,解析其原理、使用场景以及如何合理地利用线程池提高程序性能。我们将从线程池的基本概念出发,介绍其内部工作机制,然后通过实例演示如何创建和使用线程池。最后,我们将讨论线程池的优缺点以及在实际应用中需要注意的问题。
|
1天前
|
设计模式 算法 安全
Java多线程编程实战:从入门到精通
【4月更文挑战第30天】本文介绍了Java多线程编程的基础,包括线程概念、创建线程(继承`Thread`或实现`Runnable`)、线程生命周期。还讨论了线程同步与锁(同步代码块、`ReentrantLock`)、线程间通信(等待/通知、并发集合)以及实战技巧,如使用线程池、线程安全设计模式和避免死锁。性能优化方面,建议减少锁粒度和使用非阻塞算法。理解这些概念和技术对于编写高效、可靠的多线程程序至关重要。
|
1天前
|
Java 调度 开发者
Java中的多线程编程:基础知识与实践
【4月更文挑战第30天】 在现代软件开发中,多线程编程是提高程序性能和响应能力的关键。Java作为一款广泛使用的编程语言,提供了丰富的多线程支持。本文将介绍Java多线程的基础概念、实现方法以及常见问题的解决策略。我们将从线程的创建和管理入手,逐步深入到同步机制、死锁避免以及高级并发工具类的应用。通过实例代码演示和理论分析,旨在帮助读者掌握Java多线程编程的核心技能,提升软件项目的并行处理能力。
|
1天前
|
Java
java多线程售票例子
java多线程售票例子
|
1天前
|
Java 程序员
Java中的多线程编程与性能优化
【4月更文挑战第30天】本文主要探讨了Java中的多线程编程以及如何通过多线程技术来提升程序的性能。首先,我们将介绍多线程的基本概念和原理,然后深入探讨Java中实现多线程的两种主要方式:继承Thread类和实现Runnable接口。接着,我们将讨论多线程中的同步问题,包括synchronized关键字和Lock锁。最后,我们将探讨如何通过线程池来管理和优化线程,以及如何避免常见的多线程问题。
|
2天前
|
监控 安全 Java
【多线程学习】深入探究阻塞队列与生产者消费者模型和线程池常见面试题
【多线程学习】深入探究阻塞队列与生产者消费者模型和线程池常见面试题
|
2天前
|
缓存 安全 Java
多线程--深入探究多线程的重点,难点以及常考点线程安全问题
多线程--深入探究多线程的重点,难点以及常考点线程安全问题
|
2天前
|
数据采集 安全 Java
Python的多线程,守护线程,线程安全
Python的多线程,守护线程,线程安全
|
6天前
|
监控 Java 调度
Java多线程实战-从零手搓一个简易线程池(四)线程池生命周期状态流转实现
Java多线程实战-从零手搓一个简易线程池(四)线程池生命周期状态流转实现