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

目录
相关文章
|
1天前
|
安全 Java
java 中 i++ 到底是否线程安全?
本文通过实例探讨了 `i++` 在多线程环境下的线程安全性问题。首先,使用 100 个线程分别执行 10000 次 `i++` 操作,发现最终结果小于预期的 1000000,证明 `i++` 是线程不安全的。接着,介绍了两种解决方法:使用 `synchronized` 关键字加锁和使用 `AtomicInteger` 类。其中,`AtomicInteger` 通过 `CAS` 操作实现了高效的线程安全。最后,通过分析字节码和源码,解释了 `i++` 为何线程不安全以及 `AtomicInteger` 如何保证线程安全。
java 中 i++ 到底是否线程安全?
|
2天前
|
安全 Java
在 Java 中使用实现 Runnable 接口的方式创建线程
【10月更文挑战第22天】通过以上内容的介绍,相信你已经对在 Java 中如何使用实现 Runnable 接口的方式创建线程有了更深入的了解。在实际应用中,需要根据具体的需求和场景,合理选择线程创建方式,并注意线程安全、同步、通信等相关问题,以确保程序的正确性和稳定性。
|
1天前
|
缓存 Java 调度
Java中的多线程编程:从基础到实践
【10月更文挑战第24天】 本文旨在为读者提供一个关于Java多线程编程的全面指南。我们将从多线程的基本概念开始,逐步深入到Java中实现多线程的方法,包括继承Thread类、实现Runnable接口以及使用Executor框架。此外,我们还将探讨多线程编程中的常见问题和最佳实践,帮助读者在实际项目中更好地应用多线程技术。
8 3
|
3天前
|
监控 安全 Java
Java多线程编程的艺术与实践
【10月更文挑战第22天】 在现代软件开发中,多线程编程是一项不可或缺的技能。本文将深入探讨Java多线程编程的核心概念、常见问题以及最佳实践,帮助开发者掌握这一强大的工具。我们将从基础概念入手,逐步深入到高级主题,包括线程的创建与管理、同步机制、线程池的使用等。通过实际案例分析,本文旨在提供一种系统化的学习方法,使读者能够在实际项目中灵活运用多线程技术。
|
1天前
|
缓存 安全 Java
Java中的多线程编程:从基础到实践
【10月更文挑战第24天】 本文将深入探讨Java中的多线程编程,包括其基本原理、实现方式以及常见问题。我们将从简单的线程创建开始,逐步深入了解线程的生命周期、同步机制、并发工具类等高级主题。通过实际案例和代码示例,帮助读者掌握多线程编程的核心概念和技术,提高程序的性能和可靠性。
7 2
|
2天前
|
Java
Java中的多线程编程:从基础到实践
本文深入探讨Java多线程编程,首先介绍多线程的基本概念和重要性,接着详细讲解如何在Java中创建和管理线程,最后通过实例演示多线程的实际应用。文章旨在帮助读者理解多线程的核心原理,掌握基本的多线程操作,并能够在实际项目中灵活运用多线程技术。
|
2天前
|
Java
线程池内部机制:线程的保活与回收策略
【10月更文挑战第24天】 线程池是现代并发编程中管理线程资源的一种高效机制。它不仅能够复用线程,减少创建和销毁线程的开销,还能有效控制并发线程的数量,提高系统资源的利用率。本文将深入探讨线程池中线程的保活和回收机制,帮助你更好地理解和使用线程池。
17 2
|
3天前
|
Java 数据处理 开发者
Java多线程编程的艺术:从入门到精通####
【10月更文挑战第21天】 本文将深入探讨Java多线程编程的核心概念,通过生动实例和实用技巧,引导读者从基础认知迈向高效并发编程的殿堂。我们将一起揭开线程管理的神秘面纱,掌握同步机制的精髓,并学习如何在实际项目中灵活运用这些知识,以提升应用性能与响应速度。 ####
19 3
|
2天前
|
Java 开发者
Java中的多线程基础与应用
【10月更文挑战第24天】在Java的世界中,多线程是提高效率和实现并发处理的关键。本文将深入浅出地介绍如何在Java中创建和管理多线程,以及如何通过同步机制确保数据的安全性。我们将一起探索线程生命周期的奥秘,并通过实例学习如何优化多线程的性能。无论你是初学者还是有一定经验的开发者,这篇文章都将为你打开一扇通往高效编程的大门。
7 0
|
6月前
|
存储 安全 Java
深入理解Java并发编程:线程安全与锁机制
【5月更文挑战第31天】在Java并发编程中,线程安全和锁机制是两个核心概念。本文将深入探讨这两个概念,包括它们的定义、实现方式以及在实际开发中的应用。通过对线程安全和锁机制的深入理解,可以帮助我们更好地解决并发编程中的问题,提高程序的性能和稳定性。