AQS定义了一套多线程访问共享资源的同步器框架,许多同步类实现都依赖于它,如常用的ReentrantLock。简单来说,AQS定义了一套框架,来实现同步类。
2. AQS核心思想2.1 基本框架AQS的核心思想是对于共享资源,维护一个双端队列来管理线程,队列中的线程依次获取资源,获取不到的线程进入队列等待,直到资源释放,队列中的线程依次获取资源。AQS的基本框架如图所示:
(资料图片)
2.1.1 资源statestate变量表示共享资源,通常是int
类型。
getState()
、setState()
、compareAndSetState()
等。访问类型AQS定义了两种资源访问类型:独占(Exclusive):一个时间点资源只能由一个线程占用;共享(Share):一个时间点资源可以被多个线程共用。2.1.2 CLH双向队列CLH队列是一种基于逻辑队列非线程饥饿的自旋公平锁,具体介绍可参考此篇博客。CLH中每个节点都表示一个线程,处于头部的节点获取资源,而其他资源则等待。
节点结构Node
类源码如下所示:static final class Node { // 模式,分为共享与独占 // 共享模式 static final Node SHARED = new Node(); // 独占模式 static final Node EXCLUSIVE = null; // 结点状态 // CANCELLED,值为1,表示当前的线程被取消 // SIGNAL,值为-1,表示当前节点的后继节点包含的线程需要运行,也就是unpark // CONDITION,值为-2,表示当前节点在等待condition,也就是在condition队列中 // PROPAGATE,值为-3,表示当前场景下后续的acquireShared能够得以执行 // 值为0,表示当前节点在sync队列中,等待着获取锁 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; // 结点是否在共享模式下等待 final boolean isShared() { return nextWaiter == SHARED; } // 获取前驱结点,若前驱结点为空,抛出异常 final Node predecessor() throws NullPointerException { // 保存前驱结点 Node p = prev; if (p == null) // 前驱结点为空,抛出异常 throw new NullPointerException(); else // 前驱结点不为空,返回 return p; } // 无参构造方法 Node() { // Used to establish initial head or SHARED marker } // 构造方法 Node(Thread thread, Node mode) { // Used by addWaiter this.nextWaiter = mode; this.thread = thread; } // 构造方法 Node(Thread thread, int waitStatus) { // Used by Condition this.waitStatus = waitStatus; this.thread = thread; }}
Node
的方法和属性值如图所示:其中,
waitStatus
表示当前节点在队列中的状态;thread
表示当前节点表示的线程;prev
和next
分别表示当前节点的前驱节点和后继节点;nextWaiter
d当存在CONDTION队列时,表示一个condition状态的后继节点。waitStatus结点的等待状态是一个整数值,具体的参数值和含义如下所示:1
-CANCELLED
,表示节点获取锁的请求被取消,此时节点不再请求资源;0
,是节点初始化的默认值;-1
-SIGNAL
,表示线程做好准备,等待资源释放;-2
-CONDITION
,表示节点在condition等待队列中,等待被唤醒而进入同步队列;-3
-PROPAGATE
,当前线程处于共享模式下的时候会使用该字段。2.2 AQS模板AQS提供一系列结构,作为一个完整的模板,自定义的同步器只需要实现资源的获取和释放就可以,而不需要考虑底层的队列修改、状态改变等逻辑。使用AQS实现一个自定义同步器,需要实现的方法:
isHeldExclusively()
:该线程是否独占资源,在使用到condition的时候会实现这一方法;tryAcquire(int)
:独占模式获取资源的方式,成功获取返回true
,否则返回false
;tryRelease(int)
:独占模式释放资源的方式,成功获取返回true
,否则返回false
;tryAcquireShared(int)
:共享模式获取资源的方式,成功获取返回true
,否则返回false
;tryReleaseShared(int)
:共享模式释放资源的方式,成功获取返回true
,否则返回false
;一般来说,一个同步器是资源独占模式或者资源共享模式的其中之一,因此tryAcquire(int)
和tryAcquireShared(int)
只需要实现一个即可,tryRelease(int)
和tryReleaseShared(int)
同理。但是同步器也可以实现两种模式的资源获取和释放,从而实现独占和共享两种模式。
acquire(int)
是获取源码部分的顶层入口,源码如下所示:
public final void acquire(int arg) { if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) selfInterrupt();}
这段代码展现的资源获取流程如下:
tryAcquire()
尝试直接去获取资源;获取成功则直接返回如果获取失败,则addWaiter()
将该线程加入等待队列的尾部,并标记为独占模式;acquireQueued()
使线程阻塞在等待队列中获取资源,一直获取到资源后才返回。简单总结就是:
获取资源;失败就排队;排队要等待。从上文的描述可见重要的方法有三个:tryAquire()
、addWaiter()
、acquireQueued()
。下面将逐个分析其源码:
tryAcquire(int)
是获取资源的方法,源码如下所示:
protected boolean tryAcquire(int arg) { throw new UnsupportedOperationException();}
该方法是一个空方法,需要自定义同步器实现,因此在使用AQS实现同步器时,需要重写该方法。这也是“自定义的同步器只需要实现资源的获取和释放就可以”的体现。
3.1.2 addWaiter(Node.EXCLUSIVE)addWaiter(Node.EXCLUSIVE)
是将线程加入等待队列的尾部,源码如下所示:
private Node addWaiter(Node mode) { //以给定模式构造结点。mode有两种:EXCLUSIVE(独占)和SHARED(共享) //aquire()方法是独占模式,因此直接使用Exclusive参数。 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;}
首先,使用模式将当前线程构造为一个节点,然后尝试将该节点放入队尾,如果成功则返回,否则调用enq(node)
将节点放入队尾,最终返回当前节点的位置指针。其中,enq(node)
方法是将节点加入队列的方法,源码如下所示:
private Node enq(final Node node) { for (;;) { // 无限循环,确保结点能够成功入队列 // 保存尾结点 Node t = tail; if (t == null) { // 尾结点为空,即还没被初始化 if (compareAndSetHead(new Node())) // 头节点为空,并设置头节点为新生成的结点 tail = head; // 头节点与尾结点都指向同一个新生结点 } else { // 尾结点不为空,即已经被初始化过 // 将node结点的prev域连接到尾结点 node.prev = t; if (compareAndSetTail(t, node)) { // 比较结点t是否为尾结点,若是则将尾结点设置为node // 设置尾结点的next域为node t.next = node; return t; // 返回尾结点 } } }}
3.1.3 acquireQueued(Node node, int arg)这部分源码是将线程阻塞在等待队列中,线程处于等待状态,直到获取到资源后才返回,源码如下所示:
// sync队列中的结点在独占且忽略中断的模式下获取(资源)final boolean acquireQueued(final Node node, int arg) { // 标志 boolean failed = true; try { // 中断标志 boolean interrupted = false; for (;;) { // 无限循环 // 获取node节点的前驱结点 final Node p = node.predecessor(); if (p == head && tryAcquire(arg)) { // 前驱为头节点并且成功获得锁 setHead(node); // 设置头节点 p.next = null; // help GC failed = false; // 设置标志 return interrupted; } if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt())// //shouldParkAfterFailedAcquire只有当该节点的前驱结点的状态为SIGNAL时,才可以对该结点所封装的线程进行park操作。否则,将不能进行park操作。 //parkAndCheckInterrupt首先执行park操作,即禁用当前线程,然后返回该线程是否已经被中断 interrupted = true; } } finally { if (failed) cancelAcquire(node); }}
acquireQueued(Node node, int arg)
方法的主要逻辑如下:
node
节点的前驱结点,判断前驱节点是不是头部节点head,有没有成功获取资源。如果前驱结点是头部节点head并且获取了资源,说明自己应该被唤醒,设置该节点为head节点等待下一个获得资源;如果前驱节点不是头部节点或者没有获取资源,则判断是否需要park当前线程,判断前驱节点状态是不是SIGNAL
,是的话则park当前节点,否则不执行park操作;park
当前节点之后,当前节点进入等待状态,等待被其他节点unpark
操作唤醒。然后重复此逻辑步骤。3.2 release(int)release(int)
是释放资源的顶层入口方法,源码如下所示:
public final boolean release(int arg) { if (tryRelease(arg)) { // 释放成功 // 保存头节点 Node h = head; if (h != null && h.waitStatus != 0) // 头节点不为空并且头节点状态不为0 unparkSuccessor(h); //释放头节点的后继结点 return true; } return false;}
release(int)
方法的主要逻辑如下:
true
,否则返回false
;释放成功之后,需要调用unparkSuccessor(h)
唤醒后继节点。下面介绍两个重要的源码函数:tryRelease(int)
和unparkSuccessor(h)
。
tryRelease(int)
是释放资源的方法,源码如下所示:
protected boolean tryRelease(int arg) { throw new UnsupportedOperationException();}
这部分是需要自定义同步器自己实现的,要注意的是返回值需要为boolean
类型,表示释放资源是否成功。
unparkSuccessor(h)
是唤醒后继节点的方法,源码如下所示:
private void unparkSuccessor(Node node) { //这里,node一般为当前线程所在的结点。 int ws = node.waitStatus; if (ws < 0)//置零当前线程所在的结点状态,允许失败。 compareAndSetWaitStatus(node, ws, 0); Node s = node.next;//找到下一个需要唤醒的结点s 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);//唤醒}
这部分主要是查找第一个还处于等待状态的节点,将其唤醒;查找顺序是从后往前找,这是因为CLH队列中的prev
链是强一致的,从后往前找更加安全,而next
链因为addWaiter()
方法和cancelAcquire()
方法的存在,不是强一致的,因此从前往后找可能会出现问题。这部分的具体解释可以参考参考文献-1
是使用共享模式获取共享资源的顶层入口方法,源码如下所示:
public final void acquireShared(int arg) { if (tryAcquireShared(arg) < 0) doAcquireShared(arg);}
流程如下:
通过tryAcquireShared(arg)
尝试获取资源,如果获取成功则直接返回;如果获取资源失败,则调用doAcquireShared(arg)
将线程阻塞在等待队列中,直到被unpark()
/interrupt()
并成功获取到资源才返回。其中,tryAcquireShared(arg)
是获取共享资源的方法,也是需要用户自己实现。
而doAcquireShared(arg)
是将线程阻塞在等待队列中,直到获取到资源后才返回,具体流程和acquireQueued()
方法类似,源码如下所示:
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); }}
3.3.2 releaseShared(int)releaseShared(int)
是释放共享资源的顶层入口方法,源码如下所示:
public final boolean releaseShared(int arg) { if (tryReleaseShared(arg)) {//尝试释放资源 doReleaseShared();//唤醒后继结点 return true; } return false;}
流程如下:
使用tryReleaseShared(arg)
尝试释放资源,如果释放成功则返回true,否则返回false;如果释放成功,则调用doReleaseShared()
唤醒后继节点。下面介绍一下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; unparkSuccessor(h);//唤醒后继 } else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE)) continue; } if (h == head)// head发生变化 break; }}
4. 面试问题模拟Q:AQS是接口吗?有哪些没有实现的方法?看过相关源码吗?
AQS定义了一个实现同步类的框架,实现方法主要有tryAquire
和tryRelease
,表示独占模式的资源获取和释放,tryAquireShared
和tryReleaseShared
表示共享模式的资源获取和释放。源码分析如上文所述。
标签: