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

目录
相关文章
|
2月前
|
机器学习/深度学习 Java 编译器
解锁硬件潜能:Java向量化计算,性能飙升W倍!
编译优化中的机器相关优化主要包括指令选择、寄存器分配、窥孔优化等,发生在编译后端,需考虑目标平台的指令集、寄存器、SIMD支持等硬件特性。向量化计算利用SIMD技术,实现数据级并行,大幅提升性能,尤其适用于图像处理、机器学习等领域。Java通过自动向量化和显式向量API(JDK 22标准)支持该技术。
114 4
|
11天前
|
Java 大数据 Go
从混沌到秩序:Java共享内存模型如何通过显式约束驯服并发?
并发编程旨在混乱中建立秩序。本文对比Java共享内存模型与Golang消息传递模型,剖析显式同步与隐式因果的哲学差异,揭示happens-before等机制如何保障内存可见性与数据一致性,展现两大范式的深层分野。(238字)
35 4
|
14天前
|
缓存 安全 Java
如何理解Java中的并发?
Java并发指多任务交替执行,提升资源利用率与响应速度。通过线程实现,涉及线程安全、可见性、原子性等问题,需用synchronized、volatile、线程池及并发工具类解决,是高并发系统开发的关键基础。(238字)
95 4
|
20天前
|
设计模式 缓存 安全
【JUC】(6)带你了解共享模型之 享元和不可变 模型并初步带你了解并发工具 线程池Pool,文章内还有饥饿问题、设计模式之工作线程的解决于实现
JUC专栏第六篇,本文带你了解两个共享模型:享元和不可变 模型,并初步带你了解并发工具 线程池Pool,文章中还有解决饥饿问题、设计模式之工作线程的实现
82 2
|
3月前
|
Java API 调度
从阻塞到畅通:Java虚拟线程开启并发新纪元
从阻塞到畅通:Java虚拟线程开启并发新纪元
324 83
|
3月前
|
存储 Java 调度
Java虚拟线程:轻量级并发的革命性突破
Java虚拟线程:轻量级并发的革命性突破
307 83
|
5月前
|
机器学习/深度学习 消息中间件 存储
【高薪程序员必看】万字长文拆解Java并发编程!(9-2):并发工具-线程池
🌟 ​大家好,我是摘星!​ 🌟今天为大家带来的是并发编程中的强力并发工具-线程池,废话不多说让我们直接开始。
211 0
|
3月前
|
自然语言处理 Java Apache
在Java中将String字符串转换为算术表达式并计算
具体的实现逻辑需要填写在 `Tokenizer`和 `ExpressionParser`类中,这里只提供了大概的框架。在实际实现时 `Tokenizer`应该提供分词逻辑,把输入的字符串转换成Token序列。而 `ExpressionParser`应当通过递归下降的方式依次解析
265 14
|
4月前
|
Java 物联网 数据处理
Java Solon v3.2.0 史上最强性能优化版本发布 并发能力提升 700% 内存占用节省 50%
Java Solon v3.2.0 是一款性能卓越的后端开发框架,新版本并发性能提升700%,内存占用节省50%。本文将从核心特性(如事件驱动模型与内存优化)、技术方案示例(Web应用搭建与数据库集成)到实际应用案例(电商平台与物联网平台)全面解析其优势与使用方法。通过简单代码示例和真实场景展示,帮助开发者快速掌握并应用于项目中,大幅提升系统性能与资源利用率。
145 6
Java Solon v3.2.0 史上最强性能优化版本发布 并发能力提升 700% 内存占用节省 50%

热门文章

最新文章