AQS原理浅析关于Lock及AQS的一些补充:羊群效应

AQS(AbstractQueuedSynchronizer)

这个类很不容易看懂,因为它仅仅是提供了一系列公共的方法,让子类来调用。 那么要理解意思,就得从子类下手,反过来看才容易看懂。如下图所示:

AQS的子类实现

这里就以ReentrantLock排它锁为例开始讲解如何利用AQS

首先来看看ReentrantLock的构造方法

排它锁的构造方法

显然,对象中有一个属性叫sync,有两种不同的实现类

  • NonfairSync(默认实现)
  • FairSync

它们都是排它锁的内部类,不论用哪一个都能实现排它锁,只是内部可能有点原理上的区别。先以NonfairSync为例

  • lock()先通过CAS尝试将状态从0修改为1
    • 若直接修改成功,前提条件自然是锁的状态为0,则直接将线程的OWNER修改为当前线程,这是一种理想情况,如果并发粒度设置适当也是一种乐观情况。
    • 若该动作未成功,则会间接调用acquire(1)

这个acquire(int)就定义在AbstractQueuedSynchronizer

首先看tryAcquire(arg)的调用(当然传入的参数是1),在NonfairSync中,会这样来实现

首先获取这个锁的状态

  • 若状态为0,则尝试设置状态为传入的参数(这里就是1),若设置成功就代表自己获取到了锁,返回true。状态为0设置1的动作在外部就有做过一次,内部再一次做只是提升概率,而且这样的操作相对锁来讲不占开销。
  • 若状态非0,则判定当前线程是否为排它锁的Owner,如果是Owner则尝试将状态增加acquires(也就是增加1),如果这个状态值越界,则会抛出异常提示,若没有越界,将状态设置进去后返回true(实现类似于偏向的功能,可重入,但是无需进一步征用)。
  • 如果状态不是0,且自身不是owner,则返回false。

回到对acquire()的调用判定中是通过if(!tryAcquire())作为第1个条件的,如果返回true,则判定就不会成立了,自然后面的acquireQueued动作就不会再执行了,如果发生这样的情况是最理想的。

无论多么乐观,征用是必然存在的,如果征用存在则owner自然不会是自己,tryAcquire()会返回false,接着就会再调用方法:acquireQueued(addWaiter(Node.EXCLUSIVE), arg) 这个方法的调用的代码更不好懂,需要从里往外看,这里的Node.EXCLUSIVE是节点的类型,看名称应该清楚是排它类型的意思。接着调用addWaiter()来增加一个排它锁类型的节点,这个addWaiter()的代码是这样写的:

addWaiter的代码

这里创建了一个Node的对象,将当前线程和Node.EXCLUSIVE模式传入,也就是说Node节点理论上包含了这两项信息。 代码中的tail是AQS的一个属性

刚开始的时候肯定是为null,也就是不会进入第一层if判定的区域,而直接会进入enq(node)的代码,那么直接来看看enq(node)

看到了tail就应该猜到了AQS是链表,而且它还应该有一个head引用来指向链表的头节点 AQS在初始化的时候head、tail都是null,在运行时来回移动。 此时,我们最少至少知道AQS是一个基于状态(state)的链表管理方式。

enq(Node)的源码

/**
 * 这个插入会检测head tail 的初始化, 必要的话会初始化一个 dummy 节点, 这个和 ConcurrentLinkedQueue 一样的
 * Insert node into queue, initializing if necessary. See picture above.
 * @param node the node to insert
 * @return node's predecessor 返回的是前继节点
 */
/**
 * 将节点 node 加入队列
 * 这里有个注意点
 * 情况:
 *      1. 首先 queue是空的
 *      2. 初始化一个 dummy 节点
 *      3. 这时再在tail后面添加节点(这一步可能失败, 可能发生竞争被其他的线程抢占)
 *  这里为什么要加入一个 dummy 节点呢?
 *      这里的 Sync Queue 是CLH lock的一个变种, 线程节点 node 能否获取lock的判断通过其前继节点
 *      而且这里在当前节点想获取lock时通常给前继节点 打上 signal 的标识(表示当前继节点释放lock需要通知我来获取lock)
 *      若这里不清楚的同学, 请先看看 CLH lock的资料 (这是理解 AQS 的基础)
 */
private Node enq(final Node node){
    for(;;){
        Node t = tail;
        if(t == null){ // Must initialize       // 1. 队列为空 初始化一个 dummy 节点 其实和 ConcurrentLinkedQueue 一样
            if(compareAndSetHead(new Node())){  // 2. 初始化 head 与 tail (这个CAS成功后, head 就有值了, 详情将 Unsafe 操作)
                tail = head;
            }
        }else{
            node.prev = t;                      // 3. 先设置 Node.pre = pred (PS: 则当一个 node在Sync Queue里面时  node.prev 一定 != null, 但是 node.prev != null 不能说明其在 Sync Queue 里面, 因为现在的CAS可能失败 )
            if(compareAndSetTail(t, node)){     // 4. CAS node 到 tail
                t.next = node;                  // 5. CAS 成功, 将 pred.next = node (PS: 说明 node.next != null -> 则 node 一定在 Sync Queue, 但若 node 在Sync Queue 里面不一定 node.next != null)
                return t;
            }
        }
    }
}

这段代码就是链表的操作 首先这个是一个死循环,而且本身没有锁,因此可以有多个线程进来,假如某个线程进入方法,此时head、tail都是null,自然会进入if(t == null),创建一个Node,这个Node没有像开始那样给予类型和线程,很明显是一个空的Node对象

此时传入的node和某一个线程创建的Node对象

刚才我们很理想的认为只有一个线程会出现这种情况,如果有多个线程并发进入这个if判定区域,可能就会同时存在多个这样的数据结构,在各自形成数据结构后,多个线程都会去做compareAndSetHead(new Node())的动作,也就是尝试将这个临时节点设置为head,显然并发时只有一个线程会成功,因此成功的那个线程会执行tail = head,整个AQS的链表就成为

AQS被第一个请求成功的线程初始化后

有一个线程会成功修改head和tail的值,其它的线程会继续循环,再次循环会进入else 在else语句所在的逻辑中,第一步是node.prev = t,这个t就是tail的临时值,也就是首先让尝试写入的node节点的prev指针指向原来的结束节点,然后尝试通过CAS替换掉AQS中的tail的内容为当前线程的Node,无论有多少个线程并发到这里,依然只会有一个能成功,成功者执行t.next = node,也就是让原先的tail节点的next引用指向现在的node,现在的node已经成为了最新的结束节点,不成功者则会继续循环 简单使用图解的方式来说明,3个步骤如下所示

插入一个节点步骤前后动作

总之节点都是在链表尾部写入的,而且是线程安全的

知道了AQS大致的写入是一种双向链表的插入操作,但插入链表节点对锁有何用途呢,我们还得退回到addWaiter最终返回了要写入的node节点, 再回退到acquire ()中所在的代码中需要将这个返回的node节点作为acquireQueued入口参数,并传入另一个参数(依然是1),看看它里面到底做了些什么?请看下图:

/**
 * Acquires in exclusive uninterruptible mode for thread already in
 * queue. Used by condition wait methods as well as acquire
 *
 * @param node  the node
 * @param arg the acquire argument
 * @return {@code} if interrupted while waiting
 */
/**
 * 不支持中断的获取锁
 */
final boolean acquireQueued(final Node node, int arg){
    boolean failed = true;
    try {
        boolean interrupted = false;
        for(;;){
            final Node p = node.predecessor();      // 1. 获取当前节点的前继节点 (当一个n在 Sync Queue 里面, 并且没有获取 lock 的 node 的前继节点不可能是 null)
            if(p == head && tryAcquire(arg)){       // 2. 判断前继节点是否是head节点(前继节点是head, 存在两种情况 (1) 前继节点现在占用 lock (2)前继节点是个空节点, 已经释放 lock, node 现在有机会获取 lock); 则再次调用 tryAcquire尝试获取一下
                setHead(node);                       // 3. 获取 lock 成功, 直接设置 新head(原来的head可能就直接被回收)
                p.next = null; // help GC          // help gc
                failed = false;
                return interrupted;                // 4. 返回在整个获取的过程中是否被中断过 ; 但这又有什么用呢? 若整个过程中被中断过, 则最后我在 自我中断一下 (selfInterrupt), 因为外面的函数可能需要知道整个过程是否被中断过
            }
            if(shouldParkAfterFailedAcquire(p, node) && // 5. 调用 shouldParkAfterFailedAcquire 判断是否需要中断(这里可能会一开始 返回 false, 但在此进去后直接返回 true(主要和前继节点的状态是否是 signal))
                    parkAndCheckInterrupt()){      // 6. 现在lock还是被其他线程占用 那就睡一会, 返回值判断是否这次线程的唤醒是被中断唤醒
                interrupted = true;
            }
        }
    }finally {
        if(failed){                             // 7. 在整个获取中出错
            cancelAcquire(node);                // 8. 清除 node 节点(清除的过程是先给 node 打上 CANCELLED标志, 然后再删除)
        }
    }
}

这里也是一个死循环,除非进入if(p == head && tryAcquire(arg)),而p为node.predcessor()得到,这个方法返回node节点的前一个节点,也就是说只有当前一个节点是head时,进一步尝试通过tryAcquire(arg)来征用才有机会成功 tryAcquire(arg)成立的条件为:锁的状态为0,且通过CAS尝试设置状态成功或线程的持有者本身是当前线程才会返回true,我们现在来详细拆分这部分代码。 ○ 如果这个条件成功后,发生的几个动作包含: (1) 首先调用setHead(Node)的操作,这个操作内部会将传入的node节点作为AQS的head所指向的节点。线程属性设置为空(因为现在已经获取到锁,不再需要记录下这个节点所对应的线程了),再将这个节点的perv引用赋值为null。 (2) 进一步将的前一个节点的next引用赋值为null。 在进行了这样的修改后,队列的结构就变成了以下这种情况了,通过这样的方式,就可以让执行完的节点释放掉内存区域,而不是无限制增长队列,也就真正形成FIFO了:

CAS成功获取锁后,队列的变化

○ 如果这个判定条件失败 会首先判定:“shouldParkAfterFailedAcquire(p , node)”,这个方法内部会判定前一个节点的状态是否为:“Node.SIGNAL”,若是则返回true,若不是都会返回false,不过会再做一些操作:判定节点的状态是否大于0,若大于0则认为被“CANCELLED”掉了(我们没有说明几个状态的值,不过大于0的只可能被CANCELLED的状态),因此会从前一个节点开始逐步循环找到一个没有被“CANCELLED”节点,然后与这个节点的next、prev的引用相互指向;如果前一个节点的状态不是大于0的,则通过CAS尝试将状态修改为“Node.SIGNAL”,自然的如果下一轮循环的时候会返回值应该会返回true。 如果这个方法返回了true,则会执行:“parkAndCheckInterrupt()”方法,它是通过LockSupport.park(this)将当前线程挂起到WATING状态,它需要等待一个中断、unpark方法来唤醒它,通过这样一种FIFO的机制的等待,来实现了Lock的操作。

相应的,可以自己看看FairSync实现类的lock方法,其实区别不大,有些细节上的区别可能会决定某些特定场景的需求,你也可以自己按照这样的思路去实现一个自定义的锁。

接下来简单看看unlock()解除锁的方式,如果获取到了锁不释放,那自然就成了死锁,所以必须要释放,来看看它内部是如何释放的。同样从排它锁(ReentrantLock)中的unlock()方法开始

unlock方法间接调用AQS的release(1)来完成

通过tryRelease(int)方法进行了某种判定,若它成立则会将head传入到unparkSuccessor(Node)方法中并返回true,否则返回false。

首先来看看tryRelease(int)方法

就是一个设置锁状态的操作,而且是将状态减掉传入的参数值(参数是1),如果结果状态为0,就将排它锁的Owner设置为null,以使得其它的线程有机会进行执行。

在排它锁中,加锁的时候状态会增加1(当然可以自己修改这个值),在解锁的时候减掉1,同一个锁,在可以重入后,可能会被叠加为2、3、4这些值,只有unlock()的次数与lock()的次数对应才会将Owner线程设置为空,而且也只有这种情况下才会返回true。

这一点大家写代码要注意了哦,如果是在循环体中lock()或故意使用两次以上的lock(),而最终只有一次unlock(),最终可能无法释放锁。

在方法unparkSuccessor(Node)中,就意味着真正要释放锁了

它传入的是head节点(head节点是已经执行完的节点,在后面阐述这个方法的body的时候都叫head节点),内部首先会发生的动作是获取head节点的next节点, 如果获取到的节点不为空,则直接通过LockSupport.unpark()释放对应的被挂起的线程,这样一来将会有一个节点唤醒后继续进入acquireQueued的循环进一步尝试 tryAcquire()来获取锁,但是也未必能完全获取到哦,因为此时也可能有一些外部的请求正好与之征用,而且还奇迹般的成功了,那这个线程的运气就有点悲剧了,不过通常乐观认为不会每一次都那么悲剧。

再看看共享锁,从前面的排它锁可以看得出来是用一个状态来标志锁的,而共享锁也不例外,但是Java不希望去定义两个状态,所以它与排它锁的第一个区别就是在锁的状态,它用int来标志锁的状态,int有4个字节,它用高16位标志读锁(共享锁),低16位标志写锁(排它锁),高16位每次增加1相当于增加65536(通过1 << 16得到),自然的在这种读写锁中,读锁和写锁的个数都不能超过65535个(条件是每次增加1的,如果递增是跳跃的将会更少)。在计算读锁数量的时候将状态左移16位,而计算排它锁会与65535“按位求与”操作,如下图所示。

读写锁中的数量计算及限制

写锁的功能与“ReentrantLock”基本一致,区域在于它会在tryAcquire时,判定状态的时候会更加复杂一些(因此有些时候它的性能未必好)。

读锁也会写入队列,Node的类型被改为 Node.SHARED lock()时候调用的是AQS的acquireShared(int)方法,进一步调用tryAcquireShared()里面只需要检测是否有排它锁,如果没有则可以尝试通过CAS修改锁的状态,如果没有修改成功,则会自旋这个动作(可能会有很多线程在这自旋开销CPU)。如果这个自旋的过程中检测到排它锁竞争成功,那么tryAcquireShared()会返回-1,从而会走入排它锁的Node类似的流程,可能也会被park住,等待排它锁相应的线程最终调用unpark()动作来唤醒。

这就是Java提供的这种读写锁,不过这并不是共享锁的诠释,在共享锁里面也有多种机制 ,或许这种读写锁只是其中一种而已。在这种锁下面,读和写的操作本身是互斥的,但是读可以多个一起发生。

这样的锁理论上是非常适合应用读多写少的环境下(当然我们所讲的读多写少是读的比例远远大于写,而不是多一点点),理论上讲这样锁征用的粒度会大大降低,同时系统的瓶颈会减少,效率得到总体提升。

在本节中我们除了学习到AQS的内在,这个是Java层面的队列模型,其实我们也可以利用许多队列模型来解决自己的问题,甚至于可以改写模型模型来满足自己的需求

关于Lock及AQS的一些补充:

  • Lock的操作不仅仅局限于lock()/unlock(),因为这样线程可能进入WAITING状态,这个时候如果没有unpark()就没法唤醒它,可能会一直“睡”下去,可以尝试用tryLock()、tryLock(long , TimeUnit)来做一些尝试加锁或超时来满足某些特定场景的需要。 例如有些时候发现尝试加锁无法加上,先释放已经成功对其它对象添加的锁,过一小会再来尝试,这样在某些场合下可以避免“死锁”哦。
  • lockInterruptibly() 它允许抛出InterruptException,也就是当外部发起了中断操作,程序内部有可能会抛出这种异常,但是并不是绝对会抛出异常的,大家仔细看看代码便清楚了。
  • newCondition()操作,是返回一个Condition的对象,Condition只是一个接口,它要求实现await()、awaitUninterruptibly()、awaitNanos(long)、await(long , TimeUnit)、awaitUntil(Date)、signal()、signalAll()方法,AbstractQueuedSynchronizer中有一个内部类叫做ConditionObject实现了这个接口,它也是一个类似于队列的实现,具体可以参考源码。大多数情况下可以直接使用,当然觉得自己比较牛逼的话也可以参考源码自己来实现。
  • 在AQS的Node中有每个Node自己的状态(waitStatus),我们这里归纳一下,分别包含:
    • SIGNAL 从前面的代码状态转换可以看得出是前面有线程在运行,需要前面线程结束后,调用unpark()方法才能激活自己,值为:-1
    • CANCELLED 当AQS发起取消或fullyRelease()时,会是这个状态。值为1,也是几个状态中唯一一个大于0的状态,所以前面判定状态大于0就基本等价于是CANCELLED的意思。
    • CONDITION 线程基于Condition对象发生了等待,进入了相应的队列,自然也需要Condition对象来激活,值为-2。

PROPAGATE 读写锁中,当读锁最开始没有获取到操作权限,得到后会发起一个doReleaseShared()动作,内部也是一个循环,当判定后续的节点状态为0时,尝试通过CAS自旋方式将状态修改为这个状态,表示节点可以运行。 状态0 初始化状态,也代表正在尝试去获取临界资源的线程所对应的Node的状态。

羊群效应

这里说一下羊群效应,当有多个线程去竞争同一个锁的时候,假设锁被某个线程占用,那么如果有成千上万个线程在等待锁,有一种做法是同时唤醒这成千上万个线程去去竞争锁,这个时候就发生了羊群效应,海量的竞争必然造成资源的剧增和浪费,因此终究只能有一个线程竞争成功,其他线程还是要老老实实的回去等待。AQS的FIFO的等待队列给解决在锁竞争方面的羊群效应问题提供了一个思路:保持一个FIFO队列,队列每个节点只关心其前一个节点的状态,线程唤醒也只唤醒队头等待线程。其实这个思路已经被应用到了分布式锁的实践中,见:Zookeeper分布式锁的改进实现方案。

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏屈定‘s Blog

Java--CAS操作分析

CAS操作在Java中的应用很广泛,比如ConcurrentHashMap,ReentrantLock等,其常被用来解决独占锁对线程阻塞而导致的性能低下问题,是...

4713
来自专栏java系列博客

深入理解Java内存模型(五)——锁

1682
来自专栏安恒网络空间安全讲武堂

从零基础到成功解题之0ctf-ezdoor

2104
来自专栏小灰灰

Java并发学习之ReentrantLock的工作原理及使用姿势

Lock,ReentrantLock的工作原理及使用方式 jdk提供synchronized实现线程同步,但有些场景下并不灵活,如多个同步方法,每次只能有一个...

8876
来自专栏积累沉淀

Java批处理

批处理 JDBC对批处理的操作,首先简单说一下JDBC操作sql语句的简单机制。 JDBC执行数据库操作语句,首先需要将sql语句打包成为网络字...

4205
来自专栏开发技术

shiro源码篇 - shiro的session的查询、刷新、过期与删除,你值得拥有

    老公酷爱网络游戏,老婆无奈,只得告诫他:你玩就玩了,但是千万不可以在游戏里找老婆,不然,哼哼。。。     老公嘴角露出了微笑:放心吧亲爱的,我绝对不会...

2692
来自专栏JavaQ

高并发编程-ReentrantReadWriteLock深入解析

ReentrantLock在并发情况下只允许单个线程执行受保护的代码,而在大部分应用中都是读多写少,所以,如果使用ReentrantLock实现这种对共享数据的...

1153
来自专栏Linyb极客之路

并发编程之各种锁的简介

一、公平锁/非公平锁 公平锁是指多个线程按照申请锁的顺序来获取锁。 非公平锁是指多个线程获取锁的顺序并不是按照申请锁的顺序,有可能后申请的线程比先申请的线程优...

3716
来自专栏余林丰

【试验局】ReentrantLock中非公平锁与公平锁的性能测试

硬件环境:   CPU:AMD Phenom(tm) II X4 955 Processor   Memory:8G   SSD(128G):/   HDD(1...

2049
来自专栏阮一峰的网络日志

Redux 入门教程(二):中间件与异步操作

上一篇文章,我介绍了 Redux 的基本做法:用户发出 Action,Reducer 函数算出新的 State,View 重新渲染。 ? 但是,一个关键问题没有...

3374

扫码关注云+社区

领取腾讯云代金券