线程池ThreadPoolExecutor的源码中是如何解决并发问题的?

news2024/9/22 13:45:43

ThreadPoolExecutor面临哪些线程安全问题

ThreadPoolExecutor俗称线程池,作为java.util.concurrent包对外提供基础实现,以内部线程池的形式对外提供管理任务执行,线程调度,线程池管理等等服务。
然而为高效并发而生ThreadPoolExecutor自身也面临着很多线程安全问题,如果一个ThreadPoolExecutor对象被多个线程同时使用,则可能会出现以下问题:

1、如何保证线程合法的创建?
线程合法的创建指两个方面,即线程在创建完成之前,
①线程池的状态允许创建新线程,如果线程池的状态是RUNNING或者线程池的状态是shutdown但是任务队列不为空,均可以创建新线程。
②线程的总量也不得超过maximumPoolSize值。

线程池创建新线程需要经过下面的步骤:
①检查线程池的状态是否是运行;
②检查线程的数量是否小于maximumPoolSize;
③将标记线程数量的属性自增1;
④创建新线程并启动;
⑤新线程加入队列。

因为在并发环境下,线程池的状态和线程数量都在不断变化,甚至可能瞬息万变,所以完全有可能在第(1)步检查线程池状态合法之后,线程池的状态被其他用户改为STOP了,或者在进行完第(2)步检查线程数量合法之后,又有其他用户创建了新线程改变了线程数量等情况,一旦发生就会导致线程的非法创建。
要解决这个问题,最简单的方式就是使用一个synchronized关键字将从(1)到(5)步骤之间的代码全部加锁。这样确实可以保证线程安全,但是大量代码块被以悲观锁的方式加锁会带来严重的锁竞争,影响线程池的运行效率。实际上线程池在解决这个问题上要比我们想象的聪明的多。

2、如何保证线程执行任务时不被中断
线程池中的线程不能在任务执行过程中被中断,但是如果线程正在执行任务时,线程池被外部调用了线程池的shutdown()方法来终止线程池,会会将线程池中的空闲线程进行一次中断。那么正在执行任务的线程是如何避免自己被中断的呢?

3、活跃线程的数量如何确保准确
线程池ThreadPoolExecutor提供了getActiveCount();方法来获取当前活跃线程的数量。如果线程池中有A、B两个线程同时执行结束,都需要将活跃线程数量workerCount减1。那么就要进行3个步骤:
(1)从主存中读取workerCount的值
(2)对workerCount进行减1操作
(3)把workerCount重新刷新到主存
如果A线程进行第(1)和(2)步之后还没有进行第(3)步将新的值刷入主存,B线程也拿到了活跃线程数量count的值进行减1,那么就会造成两个线程结束了,count的值只减去1的后果。

注:ThreadPoolExecutor源码中使用了一个AtomicInteger型的原子整型变量“ctl”来同时记录线程池的状态和线程数量,int型数据有32位,其高3位表示线程状态,低29位表示线程数量。因此无论是线程池状态还是线程数量的改变都是针对“ctl”属性进行操作,要获取线程池状态或线程数量也要由解析“ctl”属性来获取。
这里使用workerCount只是为了方便描述,实际上在ThreadPoolExecutor源码中没有workerCount这个属性。

线程创建:乐观锁与悲观锁的完美结合

上文说到,线程创建的步骤较多,如果直接用synchronized等悲观锁的方式将线程创建的全过程加锁,将导致大量的锁竞争,在高并发环境下不适用。因此,ThreadPoolExecutor采用了乐观锁与悲观锁结合的方式来实现线程创建过程的线程安全。具体过程为我认为可以分为三大步,具体内容在线程池的addWorker(Runnable firstTask, boolean core)方法的代码中。

1、初步检查,检查线程池状态和线程数量是否允许创建新线程,检查OK后通过CAS将线程数量增加1。

①获取线程池的状态并保存。
②检查取线程池状态,如果线程池状态为stop,tidying,terminated等,或者线程池状态为shutdown并且任务队列为空,直接返回false,不允许创建线程。
③获取并检查取线程数量,如果线程数量超过额定,返回false。核心线程数不超过corePoolSize,最大线程数不超过maximumPoolSize。
④使用CAS机制将线程数量增加1,这里的CAS比较的对象是记录线程状态和线程数量的“ctl”属性。如果CAS成功,则说明在线程池状态和线程数量在低①步以后都未改变。
⑤如果CAS失败则说明线程池状态或者线程数量发生了改变。
⑥获取当前最新的线程池状态,和第①步保存的线程池状态比较,如果线程池的状态发生了改变,则回到第①步,重新从检查线程池状态开始。
⑦如果经过第⑥步发现线程池状态未变,则说明线程数量改变了,则回到第③步重新检查线程数量。
初步检查过程
由此可以看出,初步检查使用的是乐观锁机制来检查线程池状态和线程数量合法性,使用乐观锁可以有效的避免锁竞争,增加线程池的并发性能。

其对应的代码块为:

       //设置一个标志位,以便检查成功跳出两层for循环
       retry:
        //两个for循环,外层循环检查线程池状态是否运行创建新线程
        for (;;) {
            int c = ctl.get();
            //获取线程池的状态并保存
            int rs = runStateOf(c);
            //下面开始检查线程池状态是否允许创建新线程
            //如果rs >= SHUTDOWN,说明线程池的状态是为 stop,tidying,terminated
            //如果线程池处于 shutdown,且 firstTask 为 null,同时队列不为空,允许创建 worker
            //如果线程池处于stop,tidying,terminated,均不允许创建worker
            if (rs >= SHUTDOWN &&
                    ! (rs == SHUTDOWN &&
                            firstTask == null &&
                            ! workQueue.isEmpty()))
                return false;
            //内层循环检查工作线程数量是否在要求范围之内
            for (;;) {
                //获取线程数量
                int wc = workerCountOf(c);
                //如果工作线程的数量超标,直接返回false
                if (wc >= CAPACITY ||
                        wc >= (core ? corePoolSize : maximumPoolSize))
                    return false;
                //compareAndIncrementWorkerCount方法的作用是先比较ctl的当前实际值有没有改变
                //没有改变的话将ctl自增1,然后进行下一步创建线程
                if (compareAndIncrementWorkerCount(c))
                    //跳出两层for循环,执行下面的正式创建线程的代码
                    break retry;
                //CAS修改失败,说明ctl的值改变了,因为ctl的值包含了线程池状态和线程数量,
                //所以其改变就有两种可能:线程数量被并发修改了,或者线程池状态都变了
                //获取线程池最新的状态
                c = ctl.get();
                //如果线程池的状态发生改变,则需要从头开始重新获取线程池状态,
                //即重新执行外部的大循环
                if (runStateOf(c) != rs)
                    continue retry;
                    //走到这里,说明线程池状态没变,那只能是工作线程数量变了,
                    //则只需要重新检查线程数量是否合适,即重新执行内部小循环
            }
        }

2、如果初次检查和线程数量自增成功,则创建一个线程对象。这个没什么好说的,就是new一个Worker对象而已。但是需要注意的是,此时只是创建了线程实例,并没有调用线程的start()方法将线程启动。因为经过第(1)步虽然已经使用CAS完成了线程数量自增,所以不会再有线程数量超标的问题。但是线程状态仍然可能改变。

3、再次检查,创建线程对象之后,因为线程池的状态还可能会被修改,所以在线程启动之前需要使用悲观锁来上锁,防止线程池状态被修改。上锁之后再次检查线程池状态,检查通过之后启动线程,最后释放锁。
ThreadPoolExecutor源码中封装了一个ReentrantLock属性,用来给给再次检查的代码块加锁。

    /*
    * 当线程对象创建之后,线程启动之前,需要再次检查线程池状态,为了防止此期间
    * 线程池状态被修改,需要用ReentrantLock来进行加锁
    * */
    private final ReentrantLock mainLock = new ReentrantLock();

其详细流程为:
①创建线程对象
②使用ReentrantLock加锁
③检查线程池状态是否合格
④如果线程池状态OK则将线程放入队列,并将标记是否入队成功的属性workerAdded设为true。
⑤如果线程入队成功,则使用线程的start()方法启动线程。
⑥如果线程入队失败,则可能有两种原因,一是线程池状态发生改变,二是线程已经启动过了。只要线程入队失败,就要使用addWorkerFailed(Worker w)方法进行回滚。addWorkerFailed(Worker w)方法会将线程从队列中移除,并且把线程数量减1。
在这里插入图片描述
对应的代码块为:

//走到这里,说明线程池状态和线程数量都允许创建新线程
        //新线程是否已经开始运行
        boolean workerStarted = false;
        //新线程是否已经加入队列
        boolean workerAdded = false;
        Worker w = null;
        try {
            //构造器中利用线程工厂得到新线程对象
            w = new Worker(firstTask);
            //获得这个新线程
            final Thread t = w.thread;
            if (t != null) {
                final ReentrantLock mainLock = this.mainLock;
                //加锁,在新线程加入workers队列时不允许有其他线程改变线程池状态
                mainLock.lock();
                try {
                    //再次检查线程池状态,是否运行创建新的worker
                    int rs = runStateOf(ctl.get());
                    if (rs < SHUTDOWN ||
                            (rs == SHUTDOWN && firstTask == null)) {
                        //新线程肯定是还没有开始运行的,这里可能是程序员自定义ThreadFactory
                        //实现类时在内部就start了
                        if (t.isAlive())
                            throw new IllegalThreadStateException();
                        //worker加入线程队列
                        workers.add(w);
                        //调整largestPoolSize值
                        int s = workers.size();
                        if (s > largestPoolSize)
                            largestPoolSize = s;
                        workerAdded = true;
                    }
                } finally {
                    mainLock.unlock();
                }
                //加入集合成功,才启动新线程
                if (workerAdded) {
                    t.start();
                    workerStarted = true;
                }
            }
        } finally {
            //检测线程池状态发生变化后,那新线程不能被启动,则需要把已创建的Worker从集合中移除,
            // 并且把ctl的线程数量部分再减1,其实就是addWorkerFailed
            if (! workerStarted)
                addWorkerFailed(w);
        }

在上面的流程和代码中我们可以看到,ThreadPoolExecutor在创建线程的全部过程中,使用悲观锁锁住的部分仅仅只有再次检查线程池状态和线程入队列这两个步骤的几行代码,其他全部使用乐观锁来完成。并且刻意将线程创建和线程启动两个步骤细分开来,最大限度减小悲观锁的加锁区域,也最大限度减少了锁竞争的发生频率。

线程执行任务:不可重入锁防止线程被中断

线程池使用Worker来封装线程,在Worker的源码中,其继承了同步器的模板AQS并实现了一个不可重入锁。在此,先简单介绍一下AQS的实现原理:

1、AQS内部定义了一个state变量作为锁计数器,当state的值为0时,表示当前没有线程持有锁;当state的值为1时,表示有一个线程持有了锁;当state的值大于1时,表示线程重入了该锁。
2、AQS会使用一个属性来记录当前是哪个线程持有了锁。
3、AQS内部维护了一个队列,常称之为同步队列,让所有需要排队等待锁的线程都加入队列中并且进入阻塞状态。然后当持有锁的线程释放锁以后,唤醒位于队列头部的线程。
4、为了支持线程可以在某个特定的条件下等待或者唤醒,即实现Condition接口的功能,AQS内部还需要有另外一个队列,常称之为条件队列。Condition实例让一个线程进入等待状态时,该线程会被放入到条件队列。直到调用signal或signalAll方法后,再将线程转移到外部类AQS的等待队列中,线程需要获取到AQS等待队列的锁,才可以继续恢复执行后续的用户代码。
不过第4点在线程池中没有涉及到

Worker继承了AQS,并且获取锁的tryAcquire(int unused)方法设计为:使用 CAS 修改 AQS 中的锁计数器state,期望值为0(0的时候表示锁未被任何线程获取过),即只有在state为0时才能修改成功获得锁。这样就成了不可重入的独占锁。释放锁时直接将锁计数器置为0。

下面摘自Worker类的源码:

//尝试去占用当前 worker 的独占锁
        //只有锁状态原先为0并且将其改成1才能成功获取锁,这样就成了不可重入的独占锁
        protected boolean tryAcquire(int unused) {
            // 使用 CAS 修改 AQS 中的state,期望值为0(0的时候表示未被占用),
            //修改成功表示当前线程获取锁成功
            // 那么设置 ExclusiveOwnerThread 为当前线程
            if (compareAndSetState(0, 1)) {
                setExclusiveOwnerThread(Thread.currentThread());
                return true;
            }
            return false;
        }

        // 尝试释放当前 worker 的独占锁
        protected boolean tryRelease(int unused) {
            setExclusiveOwnerThread(null);
            setState(0);
            return true;
        }

在有某线程调用interruptIdleWorkers(boolean onlyOne)中断线程时,会先获取worker的独占锁才能中断worker的线程。但是正在执行任务的线程独占锁已经被占用了,而且因为该独占锁是不可重入的,所以任何线程都不能再次获取,则无法将正在执行任务的线程中断。
相关代码为:

/*
    * 此函数尝试中断空闲的Worker线程。Worker线程在执行task的前提是持有自己的Worker锁,
    * 相反,空闲的线程是没有持有自己的Worker锁的,所以当前线程执行w.tryLock()是能返回true的。
    * 参数onlyOne为true时,只中断一个空闲的Worker线程,反正中断所有空闲线程。*/
    private void interruptIdleWorkers(boolean onlyOne) {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            for (Worker w : workers) {
                Thread t = w.thread;
                //没有被中断过的线程,并且这个线程并没有在运行task(运行task时会持有worker锁)
                //注意,w.tryLock()一定得放到右边,不然可能获得锁后不释放锁
                if (!t.isInterrupted() && w.tryLock()) {
                    try {
                        t.interrupt();
                    } catch (SecurityException ignore) {
                    } finally {
                        w.unlock();
                    }
                }
                //onlyOne为true则只中断一个空闲线程即可
                if (onlyOne)
                    break;
            }
        } finally {
            mainLock.unlock();
        }
    }

线程数量的增与减:使用CAS保证准确

这个没啥好说的了,就是使用CAS机制来防止并发问题的。

    //线程数量增1
    private boolean compareAndIncrementWorkerCount(int expect) {
        return ctl.compareAndSet(expect, expect + 1);
    }
    线程数量减1
    private boolean compareAndDecrementWorkerCount(int expect) {
        return ctl.compareAndSet(expect, expect - 1);
    }

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.coloradmin.cn/o/132713.html

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈,一经查实,立即删除!

相关文章

C++项目实战:职工管理系统

1.管理系统的要求 系统可以管理公司内部所有员工的信息 主要使用c实现一个基于多态的职工管理系统 公司中的职工分为三类&#xff1a;普通员工、经理、老板&#xff0c;显示信息时需要显示职工编号、职工姓名、职工岗位以及职责 普通员工职责&#xff1a;完成经理安排的各项任…

oh my 毕设-人体姿态估计综述

文章目录What is Human Pose Estimation?Classical vs. Deep Learning-based approachesClassical approaches to 2D Human Pose EstimationDeep Learning-based approaches to 2D Human Pose EstimationHuman Pose Estimation using Deep Neural NetworksOpenPoseAlphaPose (…

想要努力赚钱,培养四种基础能力

这四种基础能力分别是&#xff1a;认知力、学习力、执行力、复盘力。我们的认知和思维&#xff0c;很大程度上&#xff0c;都是由所处的环境和圈子决定的。在同一个环境和圈子里面呆久了&#xff0c;你的认知就会被固化了。穷人最根本的枷锁&#xff0c;不是缺乏资金&#xff0…

excel图表技巧:看看,这个饼图象不象罗盘?

说到制作柱形图、条形图、饼图&#xff0c;相信大家都没有问题&#xff0c;直接选中数据&#xff0c;再插入对应的图表就行了&#xff0c;可如果要制作一张双层饼图你还会吗&#xff1f;“啥&#xff1f;还有双层饼图&#xff1f;”嘿嘿&#xff0c;不知道了吧&#xff0c;双层…

PVE+NUT+群晖等配置

文章目录配置文件说明默认配置(翻译的)ups.conf(设置ups通信相关)upsd.conf(设置ups客户访问的相关信息)upsd.users(设置upsd用户)nut.conf(nut的配置,主要是模式&#xff0c;决定使用哪些文件)upsmon.confupssched.confupssched-cmd官方手册写的可以的文章只需要实现&#xff…

excel数据查找:内容查找统计的函数公式

判断单元格是否包含特定内容是平时工作中经常会遇到的一类问题&#xff0c;常见于包含备注信息的表格中。例如下面这个考勤汇总表&#xff0c;需要根据备注中的内容判断该员工是否存在加班的情况&#xff0c;就属于这类问题。 遇到这类问题该如何处理&#xff0c;常用的公式做法…

klee2.3 教程1-2

1. klee2.3 安装 system&#xff1a;unbuntu 20.04 note: llvm-13klee2.3z3-4.10 1.1 install dependencies KLEE 需要 LLVM 的所有依赖项&#xff08;请参阅此处&#xff09;&#xff0c;以及更多。特别是&#xff0c;您应该安装下面列出的程序和库。graphviz/doxygen是可…

初级C语言之【操作符】

&#x1f996;作者&#xff1a;学写代码的恐龙 &#x1f996;博客主页&#xff1a;学写代码的恐龙博客主页 &#x1f996;专栏&#xff1a;【初级c语言】 &#x1f996;语录&#xff1a;❀未来的你&#xff0c;一定会感谢现在努力奋斗的自己❀ 初级C语言之【操作符详解】一&am…

综合能源系统分析的统一能路理论(三):《稳态与动态潮流计算》(Python代码实现)

&#x1f4a5;&#x1f4a5;&#x1f49e;&#x1f49e;欢迎来到本博客❤️❤️&#x1f4a5;&#x1f4a5; &#x1f3c6;博主优势&#xff1a;&#x1f31e;&#x1f31e;&#x1f31e;博客内容尽量做到思维缜密&#xff0c;逻辑清晰&#xff0c;为了方便读者。 ⛳️座右铭&a…

spring6笔记3(bean的循环依赖,手写spring框架,ioc注解开发,JdbcTemplate)

第九章、Bean的循环依赖问题 9.1 什么是Bean的循环依赖 A对象中有B属性。B对象中有A属性。这就是循环依赖。我依赖你&#xff0c;你也依赖我。 比如&#xff1a;丈夫类Husband&#xff0c;妻子类Wife。Husband中有Wife的引用。Wife中有Husband的引用。 public class Husband…

【java线程池详解】

java线程池详解线程的基本状态Executor框架Executor框架组成部分Executor框架使用示意图Runnable接口、Callable接口ExecutorsFuture接口和实现Future接口的FutureTask类Future和FutureTask的关系ThreadPoolExecutor类ThreadPoolExecutor 饱和策略&#xff08;拒绝策略&#xf…

MySQL去重,一条SQL语句完美解决【去重留一】

此处以某消费记录表(consume_record)为例&#xff0c;SQL语句如下&#xff1a; DELETE consume_record FROM consume_record, ( SELECT min(id) id, user_id, monetary, con…

Qt第五十五章:Qt Design Studio设计登录页并打包到python运行

目录 一、Qt Design Studio 二、导出所有文件到QRC&#xff08;不要改动默认的QRC文件名称&#xff09; 三、QRC转换成py 1.删除Constants.qml中的 2.将App.qml和Screen01.qml中的 3.转换 4、将QRC文件和转换后的py文件&#xff0c;复制到python项目中使用。 一、Qt Des…

【云原生 Kubernetes】k8s集群部署springboot项目

一、前言 本篇&#xff0c;我们将基于k8s集群&#xff0c;模拟一个比较接近实际业务的使用场景&#xff0c;使用k8s集群部署一个springboot的项目&#xff0c;我们的需求是&#xff1a; 部署SpringBoot项目到阿里云服务器 &#xff1b;基于容器打包&#xff0c;推送私有镜像仓…

Presto 之 BTreeIndex 索引代码走读

一. 前言 本文主要介绍在Presto&#xff08;OpenLookeng&#xff09;中的BTree索引的代码实现。关于BTree索引原理的介绍可以参考官网资料openLooKeng documentation。 二. BTreeIndex 索引建立 在Presto中&#xff0c;BTreeIndex 索引是通过mapdb中的BTreeMap数据结构实现的&a…

【java入门系列一】java基础

学习记录&#x1f914;写在前面JDK\JREPython有没有虚拟机&#xff1f;第一个code规范学习方法转义符号注释讨论总结谢谢点赞交流&#xff01;(❁◡❁)更多代码&#xff1a; Gitee主页&#xff1a;https://gitee.com/GZHzzz博客主页&#xff1a; CSDN&#xff1a;https://blog.…

13---SpringBoot整合JWT,实现登录和拦截

1、 JWT简介 什么是JWT&#xff1f; JWT(JSON Web Token)是为了在网络应用环境间传递声明而执行的一种基于JSON的开放标准。它将用户信息加密到token里&#xff0c;服务器不保存任何用户信息。服务器通过使用保存的密钥验证token的正确性&#xff0c;只要正确即通过验证&…

在Ubuntu上安装Azure DevOps代理程序

Contents1 概述2. 安装Ubuntu 18.04操作系统3. 安装Azure DevOps Server 代理3.1 安装Azure DevOps Server 代理3.2 以服务方式运行代理1. 概述Ubuntu是一个以桌面应用为主的Linux操作系统&#xff0c;目前在不适用微软Windows的企业中&#xff0c;ubuntu被广泛应用在个人电脑中…

网络原理4 数据链路层

文章目录mac地址网络原理的总结在数据链路层中&#xff0c;最主要的就是以太网协议这里的目的IP和原地址都是mac地址 mac地址 首先要知道什么是Mac地址&#xff0c;mac地址也叫做物理地址或以太网地址&#xff0c;它是一个用来确认网络设备位置的位置&#xff0c;一个网卡就会…

javaWeb——第一章概述

目录 1.1 软件的分类 1.2 软件架构 1.3 web软件 1.4 web程序 web服务器&#xff1a; Tomcat: 扩展 Java web就是窗口和程序之间的交互&#xff1a; 1.1 软件的分类 系统软件 应用软件 介于两者之间的中间件&#xff08;插件&#xff09; 1.2 软件架构 B/S 服务器与浏…