zoukankan      html  css  js  c++  java
  • 深入理解AQS

     

    前记

    在看JUC中并发相关的源码时经常看到AQS的身影,这到底是个什么鬼?必须要一探究竟。

    一. AQS背景了解

      JUC包中的锁,包括:

    Lock接口,ReadWriteLock接口,LockSupport阻塞原语,Condition条件,

    AbstractOwnableSynchronizer/AbstractQueuedSynchronizer/AbstractQueuedLongSynchronizer三个抽象类,

    ReentrantLock独占锁,ReentrantReadWriteLock读写锁。

    CountDownLatch,CyclicBarrier和Semaphore也是通过AQS来实现的;

       AbstractQueuedSynchronizer类,简称为AQS,通过JUC的框架图可以看出AQS是整个JUC锁的最核心的一块。了解它的原理及实现非常有助于我们去深刻的理解JUC的各种锁的实现。

     

    以ReentrantLock为例

     

     

    二.AQS涉及技术

    模板方法:模板方法模式在一个方法中定义一个算法的骨架,而将一些步骤的实现延迟到子类中。模板方法使得子类可以在不改变算法结构的情况下,重新定义算法中某些步骤的具体实现

    CHL队列:CLH也是一种基于单向链表(隐式创建)的高性能、公平的自旋锁,申请加锁的线程只需要在其前驱节点的本地变量上自旋,从而极大地减少了不必要的处理器缓存同步的次数,降低了总线和内存的开销。AQS使用的是CHL的变种。

    CAS操作:compare and swap技术 ,底层调用 Unsafe类

    LockSupport类对线程的操作支持:LockSupport.park() 阻塞线程,LockSupport.unpark() 唤醒线程,底层调用 Unsafe类 。

    Unsafe类:Unsafe是Java中一个底层类,包含了很多基础的操作,比如数组操作、对象操作、内存操作、CAS操作、线程(park)操作、栅栏(Fence)操作,JUC包、一些三方框架都使用Unsafe类来保证并发安全。其中AQS中使用到  Unsafe线程操作.许可线程通过(park),或者让线程等待许可(unpark);Unsafe的CAS操作 在Unsafe中包含compareAndSwapObject、compareAndSwapInt、compareAndSwapLong三个方法

    了解以上技术背景对于理解AQS的原理非常有帮助,这里只简要概述,根据自身需要进行深入了解或者结合后续实际分析再进行理解。

     

    三.简单认识AQS

      通过大致浏览AQS的源码先进行简单的了解AQS框架

     

    3.1 何为AQS?

    1、AQS是一个抽象类,类名为AbstractQueuedSynchronizer,抽象的都是一些公用的方法属性,其自身是没有实现任何同步接口的;

    2、AQS定义了同步器中获取锁和释放锁,目的来让自定义同步器组件来使用或重写;

    3、纵观AQS的子类,绝大多数都是一个叫Sync的静态内部类来继承AQS类,通过重写AQS中的一些方法来实现自定义同步器;

    4、AQS定义了两种资源共享方式:EXCLUSIVE( 独占式:每次仅有一个Thread能执行 )、SHARED( 共享式:多个线程可同时执行 );

    5、AQS维护了一个FIFO的CLH链表队列,且该队列不支持基于优先级的同步策略;

     

    3.2 AQS的state关键词

    1、private volatile int state:维护了一个volatile的int类型的state字段,该字段是实现AQS的核心关键词;

    2、通过getState、setState、compareAndSetState方法类获取、设置更新state值;

    3、该字段在不同的并发类中起着不同的纽带作用,下面会接着讲到state字段的一些应用场景;

     

    3.3 Node的结构

    Node,队列节点,每一个Node都持有了一个线程,对线程进行包装,方便操作。

    我们需要特别关注一下其内部的五个属性:

    static final class Node{
        /*当前node对象的等待状态,注意该状态并不是描述当前对象而是描述下一个节点的状态,
         * 从而来决定是否唤醒下一个节点,该节点总共有四个取值:
         * a. CANCELLED = 1:因为超时或者中断,结点会被设置为取消状态,被取消状态的结点不应该去竞争锁,
         * 只能保持取消状态不变,不能转换为其他状态。处于这种状态的结点会被踢出队列,被GC回收;
         * b. SIGNAL = -1:表示这个结点的继任结点被阻塞了,到时需要通知它; 
         * c. CONDITION = -2:表示这个结点在条件队列中,因为等待某个条件而被阻塞;
         * d. PROPAGATE = -3:使用在共享模式头结点有可能牌处于这种状态,表示锁的下一次获取可以无条件传播;
         * e. 0: None of the above,新结点会处于这种状态。
         * 
         * 非负值标识节点不需要被通知(唤醒)。
        */
        volatile int waitStatus;
        //当前节点的上一个节点,如果是头节点那么值为null
        volatile Node prev;
        //当前节点的下一个节点
        volatile Node next;
        //与Node绑定的线程对象
        volatile Thread thread;
        //下一个等待条件(Condition)的节点,由于Condition是独占模式,因此这里有一个简单的队列来描述Condition上的线程节点。
        Node nextWaiter;
    
    }

     

    3.4 Node的waitStatus关键词

    1、正常默认的状态值为0;

    2、对于释放操作的时候,前一个结点有唤醒后一个结点的任务;

    3、当前结点的前置结点waitStatus > 0,则结点处于CANCELLED状态,应该需要踢出队列;

    4、当前结点的前置结点waitStatus = 0,则需要将前置结点改为SIGNAL状态;
     

    3.5CLH队列


    1、队列模型:
          +------+  prev +------+  prev +------+
          |      | <---- |      | <---- |      |  
     head | Node |  next | Node |  next | Node |  tail
          |      | ----> |      | ----> |      |  
          +------+       +------+       +------+

    2、链表结构,在头尾结点中,需要特别指出的是头结点是一个空对象结点,无任何意义,即傀儡结点;
          
    3、每一个Node结点都维护了一个指向前驱的指针和指向后驱的指针,结点与结点之间相互关联构成链表;

    4、入队在尾,出队在头,出队后需要激活该出队结点的后继结点,若后继结点为空或后继结点waitStatus>0,则从队尾向前遍历取waitStatus<0的触发阻塞唤醒;


    3.6 state在AQS简单应用举例


    1、CountDownLatch,简单大致意思为:A组线程等待另外B组线程,B组线程执行完了,A组线程才可以执行;
       state初始化假设为N,后续每countDown()一次,state会CAS减1。
       等到所有子线程都执行完后(即state=0),会unpark()主调用线程,然后主调用线程就会从await()函数返回,继续后余动作。

    2、ReentrantLock,简单大致意思为:独占式锁的类;
       state初始化为0,表示未锁定状态,然后每lock()时调用tryAcquire()使state加1,
       其他线程再tryAcquire()时就会失败,直到A线程unlock()到state=0(即释放锁)为止,其它线程才有机会获取该锁;

    3、Semaphore,简单大致意思为:A、B、C、D线程同时争抢资源,目前卡槽大小为2,若A、B正在执行且未执行完,那么C、D线程在门外等着,一旦A、B有1个执行完了,那么C、D就会竞争看谁先执行;
       state初始值假设为N,后续每tryAcquire()一次,state会CAS减1,当state为0时其它线程处于等待状态,
       直到state>0且<N后,进程又可以获取到锁进行各自操作了;
     


    3.7常用重要的方法

    1、protected boolean isHeldExclusively()
       // 需要被子类实现的方法,调用该方法的线程是否持有独占锁,一般用到了condition的时候才需要实现此方法

    2、protected boolean tryAcquire(int arg)
       // 需要被子类实现的方法,独占方式尝试获取锁,获取锁成功后返回true,获取锁失败后返回false

    3、protected boolean tryRelease(int arg)  
       // 需要被子类实现的方法,独占方式尝试释放锁,释放锁成功后返回true,释放锁失败后返回false
       
    4、protected int tryAcquireShared(int arg)  
       // 需要被子类实现的方法,共享方式尝试获取锁,获取锁成功后返回正数1,获取锁失败后返回负数-1
       
    5、protected boolean tryReleaseShared(int arg)   
       // 需要被子类实现的方法,共享方式尝试释放锁,释放锁成功后返回正数1,释放锁失败后返回负数-1
       
    6、final boolean acquireQueued(final Node node, int arg)
       // 对于进入队尾的结点,检测自己可以休息了,如果可以修改则进入SIGNAL状态且进入park()阻塞状态

    7、private Node addWaiter(Node mode)
       // 添加结点到链表队尾

    8、private Node enq(final Node node)
       // 如果addWaiter尝试添加队尾失败,则再次调用enq此方法自旋将结点加入队尾

    9、private static boolean shouldParkAfterFailedAcquire(Node pred, Node node)
       // 检测结点状态,如果可以休息的话则设置waitStatus=SIGNAL并调用LockSupport.park休息;

    10、private void unparkSuccessor(Node node)   
       // 释放锁时,该方法需要负责唤醒后继节点
     

    不同的自定义同步器争用共享资源的方式也不同。自定义同步器在实现时只需要实现共享资源state的获取与释放方式即可,至于具体线程等待队列的维护(如获取资源失败入队/唤醒出队等),AQS已经在顶层实现好了。自定义同步器实现时主要实现以下几种方法:

    • isHeldExclusively():该线程是否正在独占资源。只有用到condition才需要去实现它。
    • tryAcquire(int):独占方式。尝试获取资源,成功则返回true,失败则返回false。
    • tryRelease(int):独占方式。尝试释放资源,成功则返回true,失败则返回false。
    • tryAcquireShared(int):共享方式。尝试获取资源。负数表示失败;0表示成功,但没有剩余可用资源;正数表示成功,且有剩余资源。
    • tryReleaseShared(int):共享方式。尝试释放资源,如果释放后允许唤醒后续等待结点返回true,否则返回false。

      以ReentrantLock为例,state初始化为0,表示未锁定状态。A线程lock()时,会调用tryAcquire()独占该锁并将state+1。此后,其他线程再tryAcquire()时就会失败,直到A线程unlock()到state=0(即释放锁)为止,其它线程才有机会获取该锁。当然,释放锁之前,A线程自己是可以重复获取此锁的(state会累加),这就是可重入的概念。但要注意,获取多少次就要释放多么次,这样才能保证state是能回到零态的。

      再以CountDownLatch以例,任务分为N个子线程去执行,state也初始化为N(注意N要与线程个数一致)。这N个子线程是并行执行的,每个子线程执行完后countDown()一次,state会CAS减1。等到所有子线程都执行完后(即state=0),会unpark()主调用线程,然后主调用线程就会从await()函数返回,继续后余动作。

      一般来说,自定义同步器要么是独占方法,要么是共享方式他们也只需实现tryAcquire-tryRelease、tryAcquireShared-tryReleaseShared中的一种即可但AQS也支持自定义同步器同时实现独占和共享两种方式,如ReentrantReadWriteLock。

     

    3.8设计与实现伪代码

    1、获取独占锁:
        public final void acquire(int arg) {
            if (!tryAcquire(arg) &&
                acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
                selfInterrupt();
        }

        acquire{
            如果尝试获取独占锁失败的话( 尝试获取独占锁的各种方式由AQS的子类实现 ),
            那么就新增独占锁结点通过自旋操作加入到队列中,并且根据结点中的waitStatus来决定是否调用LockSupport.park进行休息
        }
        
        
    2、释放独占锁:
        public final boolean release(int arg) {
            if (tryRelease(arg)) {
                Node h = head;
                if (h != null && h.waitStatus != 0)
                    unparkSuccessor(h);
                return true;
            }
            return false;
        }
        
        release{
            如果尝试释放独占锁成功的话( 尝试释放独占锁的各种方式由AQS的子类实现 ),
            那么取出头结点并根据结点waitStatus来决定是否有义务唤醒其后继结点
        }

    3、获取共享锁:
        public final void acquireShared(int arg) {
            if (tryAcquireShared(arg) < 0)
                doAcquireShared(arg);
        }
        
        acquireShared{
            如果尝试获取共享锁失败的话( 尝试获取共享锁的各种方式由AQS的子类实现 ),
            那么新增共享锁结点通过自旋操作加入到队尾中,并且根据结点中的waitStatus来决定是否调用LockSupport.park进行休息
        }

    4、释放共享锁:
        public final boolean releaseShared(int arg) {
            if (tryReleaseShared(arg)) {
                doReleaseShared();
                return true;
            }
            return false;
        }
        
        releaseShared{
            如果尝试释放共享锁失败的话( 尝试释放共享锁的各种方式由AQS的子类实现 ),
            那么通过自旋操作唤完成阻塞线程的唤起操作
        }

     

    四.AQS源码分析

     

     本节开始讲解AQS的源码实现。依照acquire-release、acquireShared-releaseShared的次序来。

    4.1 acquire(int)

      此方法是独占模式下线程获取共享资源的顶层入口

    如果获取到资源,线程直接返回,否则进入等待队列,直到获取到资源为止,且整个过程忽略中断的影响。这也正是lock()的语义,当然不仅仅只限于lock()。获取到资源后,线程就可以去执行其临界区代码了。下面是acquire()的源码:

    public final void acquire(int arg) {
    if (!tryAcquire(arg) &&
    acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
    selfInterrupt();
    }

     

      函数流程如下:

    1. tryAcquire()尝试直接去获取资源,如果成功则直接返回;
    2. addWaiter()将该线程加入等待队列的尾部,并标记为独占模式;
    3. acquireQueued()使线程在等待队列中获取资源,一直获取到资源后才返回。如果在整个等待过程中被中断过,则返回true,否则返回false。
    4. 如果线程在等待过程中被中断过,它是不响应的。只是获取资源后才再进行自我中断selfInterrupt(),将中断补上。

      这时单凭这4个抽象的函数来看流程还有点朦胧,不要紧,看完接下来的分析后,你就会明白了。就像《大话西游》里唐僧说的:等你明白了舍生取义的道理,你自然会回来和我唱这首歌的。

    4.2  tryAcquire(int)

      此方法尝试去获取独占资源。如果获取成功,则直接返回true,否则直接返回false。这也正是tryLock()的语义,还是那句话,当然不仅仅只限于tryLock()。如下是tryAcquire()的源码:

    protected boolean tryAcquire(int arg) {
    throw new UnsupportedOperationException();
    }

      什么?直接throw异常?说好的功能呢?好吧,还记得概述里讲的AQS只是一个框架,具体资源的获取/释放方式交由自定义同步器去实现吗?就是这里了!!!AQS这里只定义了一个接口,具体资源的获取交由自定义同步器去实现了(通过state的get/set/CAS)!!!至于能不能重入,能不能加塞,那就看具体的自定义同步器怎么去设计了!!!当然,自定义同步器在进行资源访问时要考虑线程安全的影响。

      这里之所以没有定义成abstract,是因为独占模式下只用实现tryAcquire-tryRelease,而共享模式下只用实现tryAcquireShared-tryReleaseShared。如果都定义成abstract,那么每个模式也要去实现另一模式下的接口。说到底,Doug Lea还是站在咱们开发者的角度,尽量减少不必要的工作量。

    4.3  addWaiter(Node)

      此方法用于将当前线程加入到等待队列的队尾,并返回当前线程所在的结点。还是上源码吧:

    private Node addWaiter(Node mode) {
       //以给定的模式构建结点。mode有两种:EXCLUSIVE(独占)和SHARED(共享)
    Node node = new Node(Thread.currentThread(), mode);
    //尝试快速方式直接放到队尾
    Node pred = tail;
    if (pred != null) {
    node.prev = pred;
    if (compareAndSetTail(pred, node)) {
    pred.next = node;
    return node;
    }
    }
      //上一步失败,则通过enq入队。
    enq(node);
    return node;
    }

     不用再说了,直接看注释吧。这里我们说下Node。Node结点是对每一个访问同步代码的线程的封装,其包含了需要同步的线程本身以及线程的状态,如是否被阻塞,是否等待唤醒,是否已经被取消等。变量waitStatus则表示当前被封装成Node结点的等待状态,共有4种取值CANCELLED、SIGNAL、CONDITION、PROPAGATE。

    • CANCELLED:值为1,在同步队列中等待的线程等待超时或被中断,需要从同步队列中取消该Node的结点,其结点的waitStatus为CANCELLED,即结束状态,进入该状态后的结点将不会再变化。

    • SIGNAL:值为-1,被标识为该等待唤醒状态的后继结点,当其前继结点的线程释放了同步锁或被取消,将会通知该后继结点的线程执行。说白了,就是处于唤醒状态,只要前继结点释放锁,就会通知标识为SIGNAL状态的后继结点的线程执行。

    • CONDITION:值为-2,与Condition相关,该标识的结点处于等待队列中,结点的线程等待在Condition上,当其他线程调用了Condition的signal()方法后,CONDITION状态的结点将从等待队列转移到同步队列中,等待获取同步锁。

    • PROPAGATE:值为-3,与共享模式相关,在共享模式中,该状态标识结点的线程处于可运行状态。

    • 0状态:值为0,代表初始化状态。

    AQS在判断状态时,通过用waitStatus>0表示取消状态,而waitStatus<0表示有效状态。

    4.4  enq(Node)

       此方法用于将node加入队尾。源码如下:

    private Node enq(final Node node) {
       //CAS自旋,直到成功加入队尾
    for (;;) {
    Node t = tail;
    if (t == null) { //如果队列为空,创建一个空的标志结点作为head结点,并将tail也指向它
    if (compareAndSetHead(new Node()))
    tail = head;
    } else {//正常流程放在队尾
    node.prev = t;
    if (compareAndSetTail(t, node)) {
    t.next = node;
    return t;
    }
    }
    }
    }

     

    如果你看过AtomicInteger.getAndIncrement()函数源码,那么相信你一眼便看出这段代码的精华。CAS自旋volatile变量,是一种很经典的用法。还不太了解的,自己去百度一下吧。

    4.5 acquireQueued(Node, int)

      OK,通过tryAcquire()和addWaiter(),该线程获取资源失败,已经被放入等待队列尾部了。聪明的你立刻应该能想到该线程下一部该干什么了吧:进入等待状态休息,直到其他线程彻底释放资源后唤醒自己,自己再拿到资源,然后就可以去干自己想干的事了。没错,就是这样!是不是跟医院排队拿号有点相似~~acquireQueued()就是干这件事:在等待队列中排队拿号(中间没其它事干可以休息),直到拿到号后再返回。这个函数非常关键,还是上源码吧:

    final boolean acquireQueued(final Node node, int arg) {
    boolean failed = true; //标记是否成功拿到资源
    try {
    boolean interrupted = false;
         //又是一个自旋
    for (;;) {
    final Node p = node.predecessor();//拿到前驱
           //如果前驱是head,即此结点已为老二,那么便有资格去尝试获取资源
           //可能是老大释放完资源唤醒自己的,当然也可能是被 interrrupt了。
    if (p == head && tryAcquire(arg)) {
              //拿到资源后,将head指向此结点,所以head所指的标杆结点,就是当前获得资源的那个结点或者null
    setHead(node);
    p.next = null; // setHead中node.prev已置为null,此处再将head.next置为null,就是为了方便GC回收以前的head结点。也就意味着之前拿完资源的结点出队了!
    failed = false;
    return interrupted;//返回等待过程中是否被中断过
    }
            //如果自己可以休息了,就进入waiting状态,直到被unpark()
    if (shouldParkAfterFailedAcquire(p, node) &&
    parkAndCheckInterrupt())
    interrupted = true;
    }
    } finally {
    if (failed)
    cancelAcquire(node);
    }
    }

     

    到这里了,我们先不急着总结acquireQueued()的函数流程,先看看shouldParkAfterFailedAcquire()和parkAndCheckInterrupt()具体干些什么。

    4.6 shouldParkAfterFailedAcquire(Node, Node)

      此方法主要用于检查状态,看看自己是否真的可以去休息了(进入waiting状态),万一队列前边的线程都放弃了只是瞎站着,那也说不定,对吧!

    private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    int ws = pred.waitStatus;//拿到前驱结点
    if (ws == Node.SIGNAL)
         //如果已经告诉前驱拿完号后通知自己一下,那就可以安心休息了。
    return true;
    if (ws > 0) {
         //如果前驱放弃了那就一直往前找,直到找到最近一个正常等待的状态,并排在它的后面。
         //那些被放弃的结点,由于被自己加塞到它们前面,它们相当于形成了无引用链,稍后就会被保安大叔赶走了(GC回收)
    do {
    node.prev = pred = pred.prev;
    } while (pred.waitStatus > 0);
    pred.next = node;
    } else {
         //如果前驱正常,那就把前驱的状态设置成signal,告诉它拿完号通知自己一下。
    compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
    }

     

    整个流程中,如果前驱结点的状态不是SIGNAL,那么自己就不能安心去休息,需要去找个安心的休息点,同时可以再尝试下看有没有机会轮到自己拿号。

    4.7 parkAndCheckInterrupt()

      如果线程找好安全休息点后,那就可以安心去休息了。此方法就是让线程去休息,真正进入等待状态。

    
    
    private final boolean parkAndCheckInterrupt() {
    LockSupport.park(this);//调用LockSupport.park使线程进入waiting状态
    return Thread.interrupted();//如果被唤醒查看自己是不是被中断的。
    }

       park()会让当前线程进入waiting状态。在此状态下,有两种途径可以唤醒该线程:1)被unpark();2)被interrupt()。。需要注意的是,Thread.interrupted()会清除当前线程的中断标记位。 

       

    4.8 acquireQueued小结

      OK,看了shouldParkAfterFailedAcquire()和parkAndCheckInterrupt(),现在让我们再回到acquireQueued(),总结下该函数的具体流程:

    1. 结点进入队尾后,检查状态,找到安全休息点;
    2. 调用park()进入waiting状态,等待unpark()或interrupt()唤醒自己;
    3. 被唤醒后,看自己是不是有资格能拿到号。如果拿到,head指向当前结点,并返回从入队到拿到号的整个过程中是否被中断过;如果没拿到,继续流程1。

     4.9  acquire(int)小结

     

      OKOK,acquireQueued()分析完之后,我们接下来再回到acquire()!再贴上它的源码吧:

    public final void acquire(int arg) {
    if (!tryAcquire(arg) &&
    acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
    selfInterrupt();
    }

    再来总结下它的流程吧:

    1. 调用自定义同步器的tryAcquire()尝试直接去获取资源,如果成功则直接返回;
    2. 没成功,则addWaiter()将该线程加入等待队列的尾部,并标记为独占模式;
    3. acquireQueued()使线程在等待队列中休息,有机会时(轮到自己,会被unpark())会去尝试获取资源。获取到资源后才返回。如果在整个等待过程中被中断过,则返回true,否则返回false。
    4. 如果线程在等待过程中被中断过,它是不响应的。只是获取资源后才再进行自我中断selfInterrupt(),将中断补上。

    由于此函数是重中之重,我再用流程图总结一下:

    至此,acquire()的流程终于算是告一段落了。这也就是ReentrantLock.lock()的流程,不信你去看其lock()源码吧,整个函数就是一条acquire(1)!!!

     

    4.10 release(int)

       上一小节已经把acquire()说完了,这一小节就来讲讲它的反操作release()吧。此方法是独占模式下线程释放共享资源的顶层入口。它会释放指定量的资源,如果彻底释放了(即state=0),它会唤醒等待队列里的其他线程来获取资源。这也正是unlock()的语义,当然不仅仅只限于unlock()。下面是release()的源码:

    
    
    public final boolean release(int arg) {
    if (tryRelease(arg)) {
    Node h = head; //找到头结点
    if (h != null && h.waitStatus != 0)
    unparkSuccessor(h);// 唤醒下一个结点
    return true;
    }
    return false;
    }
     

     

      逻辑并不复杂。它调用tryRelease()来释放资源。有一点需要注意的是,它是根据tryRelease()的返回值来判断该线程是否已经完成释放掉资源了!所以自定义同步器在设计tryRelease()的时候要明确这一点!!

    4.11 tryRelease(int)

      此方法尝试去释放指定量的资源。下面是tryRelease()的源码:

    protected boolean tryRelease(int arg) {
    throw new UnsupportedOperationException();
    }

     

      跟tryAcquire()一样,这个方法是需要独占模式的自定义同步器去实现的。正常来说,tryRelease()都会成功的,因为这是独占模式,该线程来释放资源,那么它肯定已经拿到独占资源了,直接减掉相应量的资源即可(state-=arg),也不需要考虑线程安全的问题。但要注意它的返回值,上面已经提到了release()是根据tryRelease()的返回值来判断该线程是否已经完成释放掉资源了!所以自义定同步器在实现时,如果已经彻底释放资源(state=0),要返回true,否则返回false。

    4.12 unparkSuccessor(Node)

      此方法用于唤醒等待队列中下一个线程。下面是源码:

    private void unparkSuccessor(Node node) {
      //这里,node一般为当前线程所在的结点
        int ws = node.waitStatus;
    if (ws < 0)//置零当前线程所在的结点状态
    compareAndSetWaitStatus(node, ws, 0);
      //找到下一个需要唤醒的结点s
    Node s = node.next;
    if (s == null || s.waitStatus > 0) {//如果为空或已取消
    s = null;
    for (Node t = tail; t != null && t != node; t = t.prev)
    if (t.waitStatus <= 0) //这里可以看出<=0的结点都为有效结点 。
    s = t;
    }
    if (s != null)
    LockSupport.unpark(s.thread);//唤醒
    }

     

      这个函数并不复杂。一句话概括:用unpark()唤醒等待队列中最前边的那个未放弃线程,这里我们也用s来表示吧。此时,再和acquireQueued()联系起来,s被唤醒后,进入if (p == head && tryAcquire(arg))的判断(即使p!=head也没关系,它会再进入shouldParkAfterFailedAcquire()寻找一个安全点。这里既然s已经是等待队列中最前边的那个未放弃线程了,那么通过shouldParkAfterFailedAcquire()的调整,s也必然会跑到head的next结点,下一次自旋p==head就成立啦),然后s把自己设置成head标杆结点,表示自己已经获取到资源了,acquire()也返回了!!And then, DO what you WANT!

    4.14  release()小结

      release()是独占模式下线程释放共享资源的顶层入口。它会释放指定量的资源,如果彻底释放了(即state=0),它会唤醒等待队列里的其他线程来获取资源。

    4.15 acquireShared(int)

      此方法是共享模式下线程获取共享资源的顶层入口。它会获取指定量的资源,获取成功则直接返回,获取失败则进入等待队列,直到获取到资源为止,整个过程忽略中断。下面是acquireShared()的源码:

    public final void acquireShared(int arg) {
    if (tryAcquireShared(arg) < 0)
    doAcquireShared(arg);
    }

     

      这里tryAcquireShared()依然需要自定义同步器去实现。但是AQS已经把其返回值的语义定义好了:负值代表获取失败;0代表获取成功,但没有剩余资源;正数表示获取成功,还有剩余资源,其他线程还可以去获取。所以这里acquireShared()的流程就是:

    1. tryAcquireShared()尝试获取资源,成功则直接返回;
    2. 失败则通过doAcquireShared()进入等待队列,直到获取到资源为止才返回。

    4.16 doAcquireShared(int)

      此方法用于将当前线程加入等待队列尾部休息,直到其他线程释放资源唤醒自己,自己成功拿到相应量的资源后才返回。下面是doAcquireShared()的源码:

    复制代码

    复制代码

    1 private void doAcquireShared(int arg) { 2 final Node node = addWaiter(Node.SHARED); 3 boolean failed = true; 4 try { 5 boolean interrupted = false;//等待过程中是否被中断过的标志 6 for (;;) { 7 final Node p = node.predecessor();//前驱 8 if (p == head) {//如果到head的下一个,因为head是拿到资源的线程,此时node被唤醒,很可能是head用完资源来唤醒自己的 9 int r = tryAcquireShared(arg);//尝试获取资源 10 if (r >= 0) {//成功 11 setHeadAndPropagate(node, r);//将head指向自己,还有剩余资源可以再唤醒之后的线程 12 p.next = null; // help GC 13 if (interrupted)//如果等待过程中被打断过,此时将中断补上。 14 selfInterrupt(); 15 failed = false; 16 return; 17 } 18 } 19 20 //判断状态,寻找安全点,进入waiting状态,等着被unpark()或interrupt() 21 if (shouldParkAfterFailedAcquire(p, node) && 22 parkAndCheckInterrupt()) 23 interrupted = true;

    private void doAcquireShared(int arg) {
    final Node node = addWaiter(Node.SHARED);//加入队列尾部
    boolean failed = true;//是否成功标志
    try {
    boolean interrupted = false;//等待过程中是否被中断过的标志
    for (;;) {
    final Node p = node.predecessor();//前驱
    if (p == head) {//如果到head的下一个,因为head是拿到资源的线程,此时node被唤醒,很可能是head用完资源来唤醒自己的
    int r = tryAcquireShared(arg);//尝试获取资源
    if (r >= 0) { //成功
    setHeadAndPropagate(node, r); //将head指向自己,还有剩余资源可以再唤醒之后的线程
    p.next = null; // help GC
    if (interrupted) //如果等待过程中被打断过,此时将中断补上
    selfInterrupt();
    failed = false;
    return;
    }
    }
            //判断状态,寻找安全点,进入waiting状态,等着被unpark()或interrupt()
    if (shouldParkAfterFailedAcquire(p, node) &&
    parkAndCheckInterrupt())
    interrupted = true;
    }
    } finally {
    if (failed)
    cancelAcquire(node);
    }
    }

     

      有木有觉得跟acquireQueued()很相似?对,其实流程并没有太大区别。只不过这里将补中断的selfInterrupt()放到doAcquireShared()里了,而独占模式是放到acquireQueued()之外,其实都一样,不知道Doug Lea是怎么想的。

      跟独占模式比,还有一点需要注意的是,这里只有线程是head.next时(“老二”),才会去尝试获取资源,有剩余的话还会唤醒之后的队友。那么问题就来了,假如老大用完后释放了5个资源,而老二需要6个,老三需要1个,老四需要2个。老大先唤醒老二,老二一看资源不够,他是把资源让给老三呢,还是不让?答案是否定的!老二会继续park()等待其他线程释放资源,也更不会去唤醒老三和老四了。独占模式,同一时刻只有一个线程去执行,这样做未尝不可;但共享模式下,多个线程是可以同时执行的,现在因为老二的资源需求量大,而把后面量小的老三和老四也都卡住了。当然,这并不是问题,只是AQS保证严格按照入队顺序唤醒罢了(保证公平,但降低了并发)。

     

    4.17  setHeadAndPropagate(Node, int)

    复制代码

    复制代码

    1 private void setHeadAndPropagate(Node node, int propagate) { 2 Node h = head; 3 setHead(node);//head指向自己 4 //如果还有剩余量,继续唤醒下一个邻居线程 5 if (propagate > 0 || h == null || h.waitStatus < 0) { 6 Node s = node.next; 7 if (s == null || s.isShared()) 8 doReleaseShared(); 9 } 10 }

    private void setHeadAndPropagate(Node node, int propagate) {
    Node h = head; // Record old head for check below
    setHead(node);//head指向自己 4
      //如果还有剩余量,继续唤醒下一个邻居线程
    if (propagate > 0 || h == null || h.waitStatus < 0 ||
    (h = head) == null || h.waitStatus < 0) {
    Node s = node.next;
    if (s == null || s.isShared())
    doReleaseShared();
    }
    }

     

      此方法在setHead()的基础上多了一步,就是自己苏醒的同时,如果条件符合(比如还有剩余资源),还会去唤醒后继结点,毕竟是共享模式!

      doReleaseShared()我们留着下一小节的releaseShared()里来讲。

     

    4.18 acquireShared小结

      OK,至此,acquireShared()也要告一段落了。让我们再梳理一下它的流程:

    1. tryAcquireShared()尝试获取资源,成功则直接返回;
    2. 失败则通过doAcquireShared()进入等待队列park(),直到被unpark()/interrupt()并成功获取到资源才返回。整个等待过程也是忽略中断的。

      其实跟acquire()的流程大同小异,只不过多了个自己拿到资源后,还会去唤醒后继队友的操作(这才是共享嘛)

    4.19 releaseShared()

      上一小节已经把acquireShared()说完了,这一小节就来讲讲它的反操作releaseShared()吧。此方法是共享模式下线程释放共享资源的顶层入口。它会释放指定量的资源,如果成功释放且允许唤醒等待线程,它会唤醒等待队列里的其他线程来获取资源。下面是releaseShared()的源码:

    public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) { //尝试释放资源
    doReleaseShared(); //唤醒后继结点
    return true;
    }
    return false;
    }

     

      此方法的流程也比较简单,一句话:释放掉资源后,唤醒后继。跟独占模式下的release()相似,

    但有一点稍微需要注意:独占模式下的tryRelease()在完全释放掉资源(state=0)后,才会返回true去唤醒其他线程,这主要是基于独占下可重入的考量

    共享模式下的releaseShared()则没有这种要求,共享模式实质就是控制一定量的线程并发执行,那么拥有资源的线程在释放掉部分资源时就可以唤醒后继等待结点。例如,资源总量是13,A(5)和B(7)分别获取到资源并发运行,C(4)来时只剩1个资源就需要等待。A在运行过程中释放掉2个资源量,然后tryReleaseShared(2)返回true唤醒C,C一看只有3个仍不够继续等待;随后B又释放2个,tryReleaseShared(2)返回true唤醒C,C一看有5个够自己用了,然后C就可以跟A和B一起运行。而ReentrantReadWriteLock读锁的tryReleaseShared()只有在完全释放掉资源(state=0)才返回true,所以自定义同步器可以根据需要决定tryReleaseShared()的返回值。

    4.20 doReleaseShared()

      此方法主要用于唤醒后继。下面是它的源码:

    private void doReleaseShared() {
    for (;;) {
    Node h = head;
    if (h != null && h != tail) {
    int ws = h.waitStatus;
    if (ws == Node.SIGNAL) {
    if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
    continue; // loop to recheck cases
    unparkSuccessor(h); //唤醒后继
    }
    else if (ws == 0 &&
    !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
    continue; // loop on failed CAS
    }
    if (h == head) // loop if head changed
    break;
    }
    }

    4.21 cquire-release、acquireShared-releaseShared 小结

      我们详解了独占和共享两种模式下获取-释放资源(acquire-release、acquireShared-releaseShared)的源码相信大家都有一定认识了。值得注意的是,acquire()和acquireShared()两种方法下,线程在等待队列中都是忽略中断的。AQS也支持响应中断的,acquireInterruptibly()/acquireSharedInterruptibly()即是,这里相应的源码跟acquire()和acquireShared()差不多,这里就不再详解了。

    4.22 LockSupport.park(),unpark()

    LockSupport是用来创建锁和其他同步类的基本线程阻塞原语。 
    LockSupport中的park() 和 unpark() 的作用分别是阻塞线程和解除阻塞线程,而且park()和unpark()不会遇到“Thread.suspend 和 Thread.resume所可能引发的死锁”问题。
    因为park() 和 unpark()有许可的存在;调用 park() 的线程和另一个试图将其 unpark() 的线程之间的竞争将保持活性。

     

    public static void park(Object blocker) {
    Thread t = Thread.currentThread();
    setBlocker(t, blocker);
    UNSAFE.park(false, 0L);
    setBlocker(t, null);
    }

     

      

       public static void unpark(Thread thread) {
            if (thread != null)
                UNSAFE.unpark(thread);
        }
    

     由原码可以看出,最终调的是sum.misc.Unsafe类中的方法

    4.23 Unsafe类

    public native void park(boolean var1, long var2);
    public native void unpark(Object var1);

    关于Usafe类,

    • Java 不能直接访问操作系统底层,而是通过本地方法来访问。Unsafe 类提供了硬件级别的原子操作。
    • Unsafe 类在 sun.misc 包下,不属于 Java 标准。很多 Java 的基础类库,包括一些被广泛使用的高性能开发库都是基于 Unsafe 类开发,比如 Netty、Hadoop、Kafka 等。
      • Unsafe 是用于在实质上扩展 Java 语言表达能力、便于在更高层(Java 层)代码里实现原本要在更低层(C 层)实现的核心库功能用的。
      • 这些功能包括裸内存的申请/释放/访问,低层硬件的 atomic/volatile 支持,创建未初始化对象等。
      • 它原本的设计就只应该被标准库使用,因此不建议在生产环境中使用。

    Usafe类功能-线程的挂起和恢复

    unpark

    • public native void unpark(Object thread);

    释放被park创建的在一个线程上的阻塞。这个方法也可以被使用来终止一个先前调用park导致的阻塞。这个操作是不安全的,因此必须保证线程是存活的(thread has not been destroyed)。从Java代码中判断一个线程是否存活的是显而易见的,但是从native代码中这机会是不可能自动完成的。

    park

    • public native void park(boolean isAbsolute, long time);

    阻塞当前线程直到一个unpark方法出现(被调用)、一个用于unpark方法已经出现过(在此park方法调用之前已经调用过)、线程被中断或者time时间到期(也就是阻塞超时)。在time非零的情况下,如果isAbsolute为true,time是相对于新纪元之后的毫秒,否则time表示纳秒。这个方法执行时也可能不合理地返回(没有具体原因)。并发包java.util.concurrent中的框架对线程的挂起操作被封装在LockSupport类中,LockSupport类中有各种版本pack方法,但最终都调用了Unsafe#park()方法。

    4.24 LCH变种队列的node

    public class Node {
        static final Node SHARED = new Node(); //模式共享模式
        static final Node EXCLUSIVE = null;   //排它模式
    static final int CANCELLED = 1; static final int SIGNAL = -1; static final int CONDITION = -2; static final int PROPAGATE = -3;
    volatile int waitStatus; //结点状态 volatile Node prev; //前驱结点引用 volatile Node next;  //后继结点引用 volatile Thread thread; //当前结点存储的线程 Node nextWaiter; private transient volatile Node head; //CLH队列 头结点 private transient volatile Node tail; //CLH队列 尾结点 private volatile int state; //结点状态
    }

      从源码看出,Node 结构的队列是一个双向链表结构,有一个head结点,一个尾结点。Node结点本身的数据部分有thread,waitStatus,模式

     

     

    五.AQS应用源码分析

     由继承关系以及类图可以看出CountDownLatch,semaphone ReentrantLock,ReentrantReadWriteLock 都有继承AQS的静态内部类Sync

     5.1 CountdownLatch

    CountDownLatch是一个同步辅助类,在完成一组正在其他线程中执行的操作之前,它允许一个或多个线程一直等待。 
      CountDownLatch的UML类图如下:

    CountDownLatch包含了sync对象,sync是Sync类型。CountDownLatch的Sync是实例类,它继承于AQS。

    countdownLatch对于同步器的静态内部类实现Sync 

        private static final class Sync extends AbstractQueuedSynchronizer {
            private static final long serialVersionUID = 4982264981922014374L;
    
            Sync(int count) {
                setState(count);  //这里调的是父类 AQS中的private vialate state 进行设置
            }
            int getCount() {
                return getState();//这里同样调的父类AQS中的private vilate state 进行取值
            }
            protected int tryAcquireShared(int acquires) {
                return (getState() == 0) ? 1 : -1;  //如果sate为0时则返回成功,否则返回失败。
           //判断 state 值是否为0,为0 返回1,否则返回 -1。state 值是 AbstractQueuedSynchronizer 类中的一个 volatile 变量。 } protected boolean tryReleaseShared(int releases) { // Decrement count; signal when transition to zero for (;;) { int c = getState(); if (c == 0) return false; int nextc = c-1; if (compareAndSetState(c, nextc))//利用CAS计数器减1. return nextc == 0; //如果计数器为0则表示已经完成了,可以通知另一个线程了,如果不为0则返回false } } }

      countdownLatch 实现的同步器sync里重写了 tryAcquireShared, tryReleaseShared,并通过 初始化传入一个count值给 AQS的violate state字段。其中state ==0是一个临界点。

    下面来看一下 CountDownLatch源码实现

    public class CountDownLatch {
    
        private static final class Sync extends AbstractQueuedSynchronizer {
            .......//此处已分析过,省略
        }
    
        private final Sync sync;
    
        public CountDownLatch(int count) {
            if (count < 0) throw new IllegalArgumentException("count < 0");
            this.sync = new Sync(count);
        }
     
        public void await() throws InterruptedException {
            sync.acquireSharedInterruptibly(1);
        }
    
    
        public boolean await(long timeout, TimeUnit unit)
            throws InterruptedException {
            return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
        }
    
        public void countDown() {
            sync.releaseShared(1);   //每调一次countdown 计数器减1.
        }
    
      
        public long getCount() {
            return sync.getCount();
        }
    
        public String toString() {
            return super.toString() + "[Count = " + sync.getCount() + "]";
        }
    }

    等待线程入队

    根据上面的逻辑,调用 await() 方法时,先去获取 state 的值,当计数器不为0的时候,说明还有需要等待的线程在运行,则调用 doAcquireSharedInterruptibly 方法,进来执行的第一个动作就是尝试加入等待队列 ,即调用 addWaiter()方法, 源码如下:

    到这里就走到了 AQS 的核心部分,AQS 用内部的一个 Node 类维护一个 CHL Node FIFO 队列。将当前线程加入等待队列,并通过 parkAndCheckInterrupt()方法实现当前线程的阻塞。下面一大部分都是在说明 CHL 队列的实现,里面用 CAS 实现队列出入不会发生阻塞。

    AQS类中的方法
    public final void acquireSharedInterruptibly(int arg) throws InterruptedException { if (Thread.interrupted()) throw new InterruptedException(); if (tryAcquireShared(arg) < 0) doAcquireSharedInterruptibly(arg); }

    等待线程被唤醒

    当执行 CountDownLatch 的 countDown()方法,将计数器减一,也就是state减一,当减到0的时候,等待队列中的线程被释放。是调用 AQS 的 releaseShared 方法来实现的,下面代码中的方法是按顺序调用的,摘到了一起,方便查看:

    // AQS类
    public final boolean releaseShared(int arg) {
            // arg 为固定值 1
            // 如果计数器state 为0 返回true,前提是调用 countDown() 之前不能已经为0
            if (tryReleaseShared(arg)) {
                // 唤醒等待队列的线程
                doReleaseShared();
                return true;
            }
            return false;
        }
    
    // CountDownLatch 重写的方法
    protected boolean tryReleaseShared(int releases) {
                // Decrement count; signal when transition to zero
                // 依然是循环+CAS配合 实现计数器减1
                for (;;) {
                    int c = getState();
                    if (c == 0)
                        return false;
                    int nextc = c-1;
                    if (compareAndSetState(c, nextc))
                        return nextc == 0;
                }
            }
    
    /// AQS类
     private void doReleaseShared() {
            for (;;) {
                Node h = head;
                if (h != null && h != tail) {
                    int ws = h.waitStatus;
                    // 如果节点状态为SIGNAL,则他的next节点也可以尝试被唤醒
                    if (ws == Node.SIGNAL) {
                        if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                            continue;            // loop to recheck cases
                        unparkSuccessor(h);
                    }
                    // 将节点状态设置为PROPAGATE,表示要向下传播,依次唤醒
                    else if (ws == 0 &&
                             !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                        continue;                // loop on failed CAS
                }
                if (h == head)                   // loop if head changed
                    break;
            }
        }
    

       5.2 Semaphore 

     Semaphore是一个计数信号量,它的本质是一个"共享锁"。
      信号量维护了一个信号量许可集。线程可以通过调用acquire()来获取信号量的许可;当信号量中有可用的许可时,线程能获取该许可;否则线程必须等待,直到有可用的许可为止。 线程可以通过release()来释放它所持有的信号量许可。
      Semaphore的UML类图如下:


      和"ReentrantLock"一样,Semaphore包含了sync对象,sync是Sync类型;而且,Sync也是一个继承于AQS的抽象类。Sync也包括"公平信号量"FairSync和"非公平信号量"NonfairSync。

    semaphone实现的同步器sync 原码

        abstract static class Sync extends AbstractQueuedSynchronizer {
            private static final long serialVersionUID = 1192457210091910933L;
    
            Sync(int permits) {
                setState(permits);// 为AQS的state字段初始化一个值:资源的总数量。
            }
    
            final int getPermits() {
                return getState();
            }
    
            final int nonfairTryAcquireShared(int acquires) {
                for (;;) {
                    int available = getState();
                    int remaining = available - acquires; //获取X个资源时,作减法,

              //1.如果剩余值小于零则说明资源不够,直接返回相减的负数。2.如果大于零,则利用CAS改变state值成功则返回正数,否则自旋至到成功。

               if (remaining < 0 ||compareAndSetState(available, remaining)) 
    return remaining; } } protected final boolean tryReleaseShared(int releases) { for (;;) { int current = getState(); int next = current + releases; //释放X个资源,作加法 if (next < current) // overflow throw new Error("Maximum permit count exceeded"); if (compareAndSetState(current, next)) //通过CAS改变State值成功则返回true,否则自旋至到成功。 return true; } }
         //减少资源,利用CAS改变 state的值。 final void reducePermits(int reductions) { for (;;) { int current = getState(); int next = current - reductions; if (next > current) // underflow throw new Error("Permit count underflow"); if (compareAndSetState(current, next)) return; } }      //对资源许可进行清零操作并返回清除的个数,也就是当前state的值。 final int drainPermits() { for (;;) { int current = getState(); if (current == 0 || compareAndSetState(current, 0)) return current; } } }

    Semaphone源码

    public class Semaphore implements java.io.Serializable {
        private static final long serialVersionUID = -3222578661600680210L;
    
        private final Sync sync;
    
        abstract static class Sync extends AbstractQueuedSynchronizer {
        }
    
        static final class NonfairSync extends Sync {
            private static final long serialVersionUID = -2694183684443567898L;
    
            NonfairSync(int permits) {
                super(permits);
            }
    
            protected int tryAcquireShared(int acquires) {
                return nonfairTryAcquireShared(acquires);
            }
        }
    
        static final class FairSync extends Sync {
            private static final long serialVersionUID = 2014338818796000944L;
    
            FairSync(int permits) {
                super(permits);
            }
    
            protected int tryAcquireShared(int acquires) {
                for (;;) {
                    if (hasQueuedPredecessors()) //公平锁需判断队列中是否有前驱节点,如果有则返回负数,没有才进行减法操作并 通过CAS改变state值而获得锁。
                        return -1;
                    int available = getState();
                    int remaining = available - acquires;
                    if (remaining < 0 ||
                        compareAndSetState(available, remaining))
                        return remaining;
                }
            }
        }
    
      
        public Semaphore(int permits) {
            sync = new NonfairSync(permits);
        }
    
       
        public Semaphore(int permits, boolean fair) {
            sync = fair ? new FairSync(permits) : new NonfairSync(permits);
        }
    
    
        public void acquire() throws InterruptedException {
            sync.acquireSharedInterruptibly(1);
        }
    
       
        public void acquireUninterruptibly() {
            sync.acquireShared(1);
        }
    
        public boolean tryAcquire() {
            return sync.nonfairTryAcquireShared(1) >= 0;
        }
    
        public boolean tryAcquire(long timeout, TimeUnit unit)
            throws InterruptedException {
            return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
        }
    
       
        public void release() {
            sync.releaseShared(1);
        }
    
        public void acquire(int permits) throws InterruptedException {
            if (permits < 0) throw new IllegalArgumentException();
            sync.acquireSharedInterruptibly(permits);
        }
    
        public void acquireUninterruptibly(int permits) {
            if (permits < 0) throw new IllegalArgumentException();
            sync.acquireShared(permits);
        }
    
        public boolean tryAcquire(int permits) {
            if (permits < 0) throw new IllegalArgumentException();
            return sync.nonfairTryAcquireShared(permits) >= 0;
        }
    
        public boolean tryAcquire(int permits, long timeout, TimeUnit unit)
            throws InterruptedException {
            if (permits < 0) throw new IllegalArgumentException();
            return sync.tryAcquireSharedNanos(permits, unit.toNanos(timeout));
        }
    
        public void release(int permits) {
            if (permits < 0) throw new IllegalArgumentException();
            sync.releaseShared(permits);
        }
    
        public int availablePermits() {
            return sync.getPermits();
        }
    
        public int drainPermits() {
            return sync.drainPermits();
        }
    
        protected void reducePermits(int reduction) {
            if (reduction < 0) throw new IllegalArgumentException();
            sync.reducePermits(reduction);
        }
    
        public boolean isFair() {
            return sync instanceof FairSync;
        }
    
    
        public final boolean hasQueuedThreads() {
            return sync.hasQueuedThreads();
        }
    
    
        public final int getQueueLength() {
            return sync.getQueueLength();
        }
    
    
        protected Collection<Thread> getQueuedThreads() {
            return sync.getQueuedThreads();
        }
    
        public String toString() {
            return super.toString() + "[Permits = " + sync.getPermits() + "]";
        }
    

     5.3 ReentrantLock

    ReentrantLock是独占锁。所谓独占锁,是指只能被独自占领,即同一个时间点只能被一个线程锁获取到的锁。ReentrantLock锁包括"公平的ReentrantLock"和"非公平的ReentrantLock"。"公平的ReentrantLock"是指"不同线程获取锁的机制是公平的",而"非公平的  ReentrantLock"则是指"不同线程获取锁的机制是非公平的",ReentrantLock是"可重入的锁"。
      ReentrantLock的UML类图如下:

      (01) ReentrantLock实现了Lock接口。
      (02) ReentrantLock中有一个成员变量sync,sync是Sync类型;Sync是一个抽象类,而且它继承于AQS。
      (03) ReentrantLock中有"公平锁类"FairSync和"非公平锁类"NonfairSync,它们都是Sync的子类。ReentrantReadWriteLock中sync对象,是FairSync与NonfairSync中的一种,这也意味着ReentrantLock是"公平锁"或"非公平锁"中的一种,ReentrantLock默认是非公平锁。

     ReentrantLock内部静态类同步器的实现 Sync

        abstract static class Sync extends AbstractQueuedSynchronizer {
            private static final long serialVersionUID = -5179523762034025860L;
            abstract void lock();
    
            final boolean nonfairTryAcquire(int acquires) {
                final Thread current = Thread.currentThread();
                int c = getState();
                if (c == 0) {
                    if (compareAndSetState(0, acquires)) { //如果state状态为0,则利用CAS自旋设置state。
                        setExclusiveOwnerThread(current); //将当前线程对象 设置给AQS的父类中的exclusiveOwnerThread字段。
                        return true;
                    }
                }
                else if (current == getExclusiveOwnerThread()) {
                    int nextc = c + acquires;
                    if (nextc < 0) // overflow
                        throw new Error("Maximum lock count exceeded");
                    setState(nextc);
                    return true;
                }
                return false;
            }
    
            protected final boolean tryRelease(int releases) {
                int c = getState() - releases;
                if (Thread.currentThread() != getExclusiveOwnerThread())
                    throw new IllegalMonitorStateException();
                boolean free = false;
                if (c == 0) { //如果状态为0则表示当前线程已释放锁。
                    free = true;
                    setExclusiveOwnerThread(null);
                }
                setState(c);
                return free;
            }
    
            protected final boolean isHeldExclusively() {
                // While we must in general read state before owner,
                // we don't need to do so to check if current thread is owner
                return getExclusiveOwnerThread() == Thread.currentThread();
            }
    
            final ConditionObject newCondition() {
                return new ConditionObject();
            }
    
            // Methods relayed from outer class
    
            final Thread getOwner() {
                return getState() == 0 ? null : getExclusiveOwnerThread();
            }
    
            final int getHoldCount() {
                return isHeldExclusively() ? getState() : 0;
            }
    
            final boolean isLocked() {
                return getState() != 0;
            }
            private void readObject(java.io.ObjectInputStream s)
                throws java.io.IOException, ClassNotFoundException {
                s.defaultReadObject();
                setState(0); // reset to unlocked state
            }
        }
    

     公平锁实现:

        static final class FairSync extends Sync {
            private static final long serialVersionUID = -3000897897090466540L;
    
            final void lock() {
                acquire(1);
            }
    
            protected final boolean tryAcquire(int acquires) {
                final Thread current = Thread.currentThread();
                int c = getState();
                if (c == 0) {//state状态为0,才会继续获取锁。判断没有后继者的情况
                    if (!hasQueuedPredecessors() &&
                        compareAndSetState(0, acquires)) {
                        setExclusiveOwnerThread(current);
                        return true;
                    }
                }
           //state不为0,则判断如果当前线程 是和当前占用锁的线程为同一个,则允许多次获取,这里是可重入锁的概念 else if (current == getExclusiveOwnerThread()) { int nextc = c + acquires; if (nextc < 0) throw new Error("Maximum lock count exceeded"); setState(nextc); return true; } return false; } }

      非公平锁实现

       static final class NonfairSync extends Sync {
            private static final long serialVersionUID = 7316153563782823691L;
    
            final void lock() {
                if (compareAndSetState(0, 1))
                    setExclusiveOwnerThread(Thread.currentThread());
                else
                    acquire(1);
            }
    
            protected final boolean tryAcquire(int acquires) {
                return nonfairTryAcquire(acquires);
            }
        }
    

      ReentrantLock源码

    public class ReentrantLock implements Lock, java.io.Serializable {
        private static final long serialVersionUID = 7373984872572414699L;
    
        private final Sync sync;
    
        abstract static class Sync extends AbstractQueuedSynchronizer {
               //。。。此处省略
        }
    
        static final class NonfairSync extends Sync {
                //。。。此处省略
        }
    
        static final class FairSync extends Sync {
                 //。。。此处省略
        }
    
        public ReentrantLock() {
            sync = new NonfairSync();
        }
    
        public ReentrantLock(boolean fair) {
            sync = fair ? new FairSync() : new NonfairSync();
        }
    
        public void lock() {
            sync.lock();
        }
    
    
        public void lockInterruptibly() throws InterruptedException {
            sync.acquireInterruptibly(1);
        }
    
        public boolean tryLock() {
            return sync.nonfairTryAcquire(1);
        }
    
        public boolean tryLock(long timeout, TimeUnit unit)
                throws InterruptedException {
            return sync.tryAcquireNanos(1, unit.toNanos(timeout));
        }
    
    
        public void unlock() {
            sync.release(1);
        }
    
     
        public Condition newCondition() {
            return sync.newCondition();
        }
    
        public int getHoldCount() {
            return sync.getHoldCount();
        }
    
        public boolean isHeldByCurrentThread() {
            return sync.isHeldExclusively();
        }
    
    
        public boolean isLocked() {
            return sync.isLocked();
        }
    
    
        public final boolean isFair() {
            return sync instanceof FairSync;
        }
    
        protected Thread getOwner() {
            return sync.getOwner();
        }
    
     
        public final boolean hasQueuedThreads() {
            return sync.hasQueuedThreads();
        }
        public final boolean hasQueuedThread(Thread thread) {
            return sync.isQueued(thread);
        }
        public final int getQueueLength() {
            return sync.getQueueLength();
        }
        protected Collection<Thread> getQueuedThreads() {
            return sync.getQueuedThreads();
        }
    
        public boolean hasWaiters(Condition condition) {
            if (condition == null)
                throw new NullPointerException();
            if (!(condition instanceof AbstractQueuedSynchronizer.ConditionObject))
                throw new IllegalArgumentException("not owner");
            return sync.hasWaiters((AbstractQueuedSynchronizer.ConditionObject)condition);
        }
    
     
        public int getWaitQueueLength(Condition condition) {
            if (condition == null)
                throw new NullPointerException();
            if (!(condition instanceof AbstractQueuedSynchronizer.ConditionObject))
                throw new IllegalArgumentException("not owner");
            return sync.getWaitQueueLength((AbstractQueuedSynchronizer.ConditionObject)condition);
        }
    
     
        protected Collection<Thread> getWaitingThreads(Condition condition) {
            if (condition == null)
                throw new NullPointerException();
            if (!(condition instanceof AbstractQueuedSynchronizer.ConditionObject))
                throw new IllegalArgumentException("not owner");
            return sync.getWaitingThreads((AbstractQueuedSynchronizer.ConditionObject)condition);
        }
    
      
        public String toString() {
            Thread o = sync.getOwner();
            return super.toString() + ((o == null) ?
                                       "[Unlocked]" :
                                       "[Locked by thread " + o.getName() + "]");
        }
    }
    

      

  • 相关阅读:
    jenkins+ant+jmeter实现自动化集成(详解)
    Android ADB命令 adb devices 出现error:protocol fault (no status)
    使用jmeter做接口测试
    估计新版本所需要的案例数量
    《将博客搬至CSDN》
    HDFS NN refreshNodes操作的可用性和效率的改进
    visudo: /etc/sudoers 忙,请稍后重试解决办法
    Java try与finally对返回值的影响
    Java中基本类型和包装类比较的问题
    Java中ArrayList去重方法
  • 原文地址:https://www.cnblogs.com/shoshana-kong/p/10835181.html
Copyright © 2011-2022 走看看