深入理解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() + "]");
    }
}

  

原文地址:https://www.cnblogs.com/shoshana-kong/p/10835181.html