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爬虫的瓶颈分析与动态线程优化方案
Java 数据库 Spring
59 0
|
1月前
|
算法 Java
Java多线程编程:实现线程间数据共享机制
以上就是Java中几种主要处理多线程序列化资源以及协调各自独立运行但需相互配合以完成任务threads 的技术手段与策略。正确应用上述技术将大大增强你程序稳定性与效率同时也降低bug出现率因此深刻理解每项技术背后理论至关重要.
86 16
|
2月前
|
缓存 并行计算 安全
关于Java多线程详解
本文深入讲解Java多线程编程,涵盖基础概念、线程创建与管理、同步机制、并发工具类、线程池、线程安全集合、实战案例及常见问题解决方案,助你掌握高性能并发编程技巧,应对多线程开发中的挑战。
|
2月前
|
数据采集 存储 前端开发
Java爬虫性能优化:多线程抓取JSP动态数据实践
Java爬虫性能优化:多线程抓取JSP动态数据实践
|
3月前
|
数据采集 监控 调度
干货分享“用 多线程 爬取数据”:单线程 + 协程的效率反超 3 倍,这才是 Python 异步的正确打开方式
在 Python 爬虫中,多线程因 GIL 和切换开销效率低下,而协程通过用户态调度实现高并发,大幅提升爬取效率。本文详解协程原理、实战对比多线程性能,并提供最佳实践,助你掌握异步爬虫核心技术。
|
3月前
|
安全 算法 Java
Java 多线程:线程安全与同步控制的深度解析
本文介绍了 Java 多线程开发的关键技术,涵盖线程的创建与启动、线程安全问题及其解决方案,包括 synchronized 关键字、原子类和线程间通信机制。通过示例代码讲解了多线程编程中的常见问题与优化方法,帮助开发者提升程序性能与稳定性。
144 0
|
4月前
|
Java 数据挖掘 调度
Java 多线程创建零基础入门新手指南:从零开始全面学习多线程创建方法
本文从零基础角度出发,深入浅出地讲解Java多线程的创建方式。内容涵盖继承`Thread`类、实现`Runnable`接口、使用`Callable`和`Future`接口以及线程池的创建与管理等核心知识点。通过代码示例与应用场景分析,帮助读者理解每种方式的特点及适用场景,理论结合实践,轻松掌握Java多线程编程 essentials。
250 5
|
8月前
|
Python
python3多线程中使用线程睡眠
本文详细介绍了Python3多线程编程中使用线程睡眠的基本方法和应用场景。通过 `time.sleep()`函数,可以使线程暂停执行一段指定的时间,从而控制线程的执行节奏。通过实际示例演示了如何在多线程中使用线程睡眠来实现计数器和下载器功能。希望本文能帮助您更好地理解和应用Python多线程编程,提高程序的并发能力和执行效率。
258 20
|
8月前
|
安全 Java C#
Unity多线程使用(线程池)
在C#中使用线程池需引用`System.Threading`。创建单个线程时,务必在Unity程序停止前关闭线程(如使用`Thread.Abort()`),否则可能导致崩溃。示例代码展示了如何创建和管理线程,确保在线程中执行任务并在主线程中处理结果。完整代码包括线程池队列、主线程检查及线程安全的操作队列管理,确保多线程操作的稳定性和安全性。

热门文章

最新文章