源码阅读笔记：AbstractQueuedSynchronizer

锁是面向使用者的，定义了使用者与锁交互的接口，隐藏了实现细节；同步器是面向所得实现者的，它简化了锁的实现方式，屏蔽了同步状态管理、线程的排队、等待与唤醒等底层操作。1

AbstractQueuedSynchronizer（以下简称AQS），就是所谓的同步器。更准确地说是队列同步器，描述了线程在获取共享资源时以FIFO的方式依次获取同步状态的一种同步模型。这个模型扩展了CLH同步模型，使得其适用于大部分线程同步场景。《Java并发编程的艺术》对AQS进行了详尽的介绍，下面将从CLH出发阐述一下自己对AQS的理解。

与CLH的异同

单从AQS源码和上一篇CLH文章的分析上看，AQS和CLH主要的区别如下：

CLH是一个单向链表。每个节点通过自旋观察前驱节点是否不再征用锁的状态来判断自己能获得锁。这个自旋过程实际上会阻塞线程继续执行。每个线程依据获取锁的先后顺序，依次等待所有先获取锁线程征用并完成执行，并修改自身标志位（是一个volatile变量，标识自己是否不再征用锁）——这个用节点的角度来描述即是：后继节点获取锁前，主动等待前驱节点释放锁。

AQS是一个双向链表。并且基于一个事实，简化了同步状态的获取——头节点是当前拥有同步状态的节点，而只有头节点的后继有资格去尝试获取同步状态。因此在AQS中，同步队列里的所有线程都在自省地观察自己的前驱节点是不是头节点，这里同样蕴含着一个volatile读操作，这点与CLH并无不同，但是这个过程已经将线程间的通信降低到最小程度。下面是独占模式下线程的自旋观察过程：

final boolean acquireQueued(final Node node, int arg) {
boolean failed = true;
try {
boolean interrupted = false;
//自旋观察自己的前驱节点
for (;;) {
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())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}

除此之外，AQS不再限于CLH这样独占式、阻塞地获取同步态的过程。AQS提供了

tryAcquire

用以非阻塞地获取同步态，以及

tryAcquireShared

用于共享式地获取同步态。但是这两个方法都没给出具体实现，实现者也可以把

tryAcquire

设计成阻塞地获取锁：

protected boolean tryAcquire(int arg) {
for(;;) {
// 调用tryAcquired的线程将一直自旋，直到CAS设置成功，返回占有锁
int current = getState();
if(current == 0 && compareAndSetState(0,1)) {
return true;
}
}
}

而从根本上说，CLH与AQS是基于同一个模式实现的：

votatile变量维护一系列共享资源：包括但不限于同步器状态、同步队列的头尾节点以及节点的同步状态

通过CAS操作完成对竞态资源的修改

通过自旋、等待与唤醒等机制，完成锁的顺序获取

AQS的基本实现

CLH算法本质完成了一个阻塞独占式的同步状态获取过程。它没有什么缺陷，只是看是不是在一个场景下依旧适用。CLH的自旋过程中，同步队列中所有的线程都在自旋。这个决定了当有大量线程征用同步态，而每个线程持有同步态的时间又很长时，CPU要分配大量的时间片给线程做无意义的自旋。类似的情况，在JDK对synchronize关键字进行优化的时候也有出现。JDK就是考虑到悲观锁频繁地改变线程的状态，所以引入了基于CAS以及自旋的乐观锁，即便如此JVM也会对自旋的次数进行限制。

一般来说，我们实现一个锁，通常会这样写：

public class MyLock implements Lock {
private final AbstractQueuedSynchronizer sync = new Sync();

private class Sync extends AbstractQueuedSynchronizer {

@Override
protected boolean tryAcquire(int arg) {
//覆写获取同步状态的方法
//通常会用AQS提供的compareAndSetState方法改写AQS的state字段
}
@Override
protected boolean tryRelease() {
//覆写释放同步状态的方法
}
}

public void lock() {
sync.acquire(1);
}

public void unlock() {
sync.release(1);
}
}

AQS做出了更加通用的实现，为锁的实现带来了极大的方便。总结一下AQS的整体结构。

AbstractQueuedSynchronizer.state

state是AQS中的一个私有变量（被声明为volatile的int），通常用来标识一个同步器的状态。例如，独占锁可以理解为占用或是空闲重入锁，重入锁则可以用它来记录重入的次数，共享锁则可以用它来记录被共享次数，读写锁则可以将这个状态拆成高低位两部分分别记录读/写状态（参考ReentrantReadWriteLock）。因此，state字段怎么使用，完全看锁的实现者如何实现

tryAcquire

和

4000
tryAcquireShared

等同步状态获取的方法。

AbstractQueuedSynchronizer.Node.waitStatus

Node是AQS定义的一个内部类，其将线程包装成了一个同步节点，从而组成同步队里。后文将会提到AQS中的等待队列，其节点也同样是Node类型的。与上面的同步状态state不同的是，waitStatus表示的含义已经明确给出2。

状态	值	含义
CANCELLED	1	由于在同步队列中等待超时或被终端，需要从同步队列中取消等待，是一个终态
SIGNAL	-1	后继节点的线程处于等待状态，而当前节点的线程如果释放了同步状态或者被取消，将会通知后继节点
CONDITION	-2	节点在等待队列中，等待在Condition上，当其他线程调用signal()方法后，该节点会从等待队列中转移到同步队列中
PROPAGATE	-3	表示下一次共享式的同步状态获取将会无条件地被传播下去
	0	非以上任意状态

同步队列

FIFO的双向同步队列是AQS的核心，也是前文跟CLH做对比的原因。实际上，严格基于FIFO的同步队列实现的锁，一定是公平锁。所谓的公平锁，就是严格按照获取锁的先后顺序去获取同步状态。下面是同步队列的示意图。



但是，AQS中的队列真的是公平的吗？如果这样，基于AQS的ReentranceLock如何实现非公平锁的？先放结论：

AQS的同步队列是公平的，但是基于AQS实现公平锁需要实现者作额外保证。

带着这个疑问，回顾一下

acquireQueued

，这个方法是所有独占锁必经的自旋过程（只要你想通过调用

acquire

来实现

lock()

）。

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

final boolean acquireQueued(final Node node, int arg) {
boolean failed = true;
try {
boolean interrupted = false;
//自旋观察自己的前驱节点
for (;;) {
final Node p = node.predecessor();
//当前驱结点是头结点，也即同步状态的占有节点时
//尝试获取同步态
if (p == head && tryAcquire(arg)) {
//获得同步态，取代自己的前驱成为头结点
//return
}
//前驱节点不是头节点时
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
...
}
}

我想，如果自旋持续进行，那么整个同步队列一定是FIFO的。因为只有前驱节点是头结点也即占有同步状态的节点时，才能去尝试获取同步状态，否则该线程会继续自旋。那么是不是所有线程都在持续地自旋呢？我想，

parkAndCheckInterrupt

应该是所有疑问的根源。

private final boolean parkAndCheckInterrupt() {
LockSupport.park(this);
return Thread.interrupted();
}

这个方法很简单，调用了

LockSupport.park(this)

，会阻塞当前线程。那么就可以理解了，一个线程可以会放弃自旋而转而选择被阻塞，直到调用

LockSupport.unpark(Thread thread)

而被唤醒。那么，唤醒的过程，是否就会导致有些线程不再遵守公平锁或是FIFO的规则？找了一下调用

LockSupport.unpark

的方法，最终由

unparkSuccessor

进行调用，而

unparkSuccessor

这个函数主要供

release

和

releaseShared

两个方法使用，用于通知头结点的后继节点，表明头结点释放了同步状态，其后继节点可以开始运行。

public final boolean release(int arg) {
// 调用tryRelease释放同步状态state
if (tryRelease(arg)) {
// 释放后通知后继节点unpark
Node h = head;
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}

那么就可以理解了——当一个节点释放同步状态时，首先释放了同步状态，然后再通知后继节点可以恢复运行，继续

acquireQueued

中的自旋过程。然而，此时随便来了一个新的线程A，完全可以抢先占领同步状态，因为

acquire

方法中，线程会先尝试获取同步状态，失败后才会构造节点进入整个同步队列。下面是ReentranceLock.FairSync中的同步状态获取方法，可以看到有一个

hasQueuedPredecessors

方法，如果线程A发现有先获取同步状态的线程（除其本身），就会直接返回。这样就把趁着间隙抢占同步状态的线程A挡在了外面。

protected final boolean tryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
if (c == 0) {
if (!hasQueuedPredecessors() && compareAndSetState(0, acquires)) {
setExclusiveOwnerThread(current);
return true;
}
}
...
}

下面是独占锁中线程获取同步状态的流程图，图中标出了各个步骤对应的方法，可以结合上面的源代码来理解。蓝色的部分是整个同步队列的处理流程，毋庸置疑其是FIFO的；红色的部分就是上面讨论的线程“插队”的根源，为自由线程提供了一个“绿色通道”，也是“非公平”的成因。



上述流程图省略了线程进入阻塞状态的前提，就是需要前驱节点的waitStatus被设置为Node.SINGAL。只有这样前驱节点在release时才会唤醒当前节点的线程。

shouldParkAfterFailedAcquire

就是完成这样的过程——这个过程中，节点会先将状态Node.CANCEL的前驱节点全部移除同步队列，之后用CAS将前驱节点设置成Node.SINGAL，然后安然地进入阻塞状态。

另外有一个很有意思的地方，

shouldParkAfterFailedAcquire

这个方法中，并不会直接将前驱节点置为Node.SINGAL并返回true，而是在下一次循环中，再返回true。注释中的解释是为了再次确认节点不能获取到同步状态，可以阻塞：

waitStatus must be 0 or PROPAGATE. Indicate that we need a signal, but don’t park yet. Caller will need to retry to make sure it cannot acquire before parking.

等待队列

熟悉synchronized的同学，应该都记得其通过监视器提供了

notify

和

wait

的等待-通知模型。使线程可以通过synchronized完成更细粒度的同步。同样，AQS也提供了类似的方法，甚至更加强大。



AQS内有一个ConditionObject子类，每个子类可以视为一个等待队列，如上图所示。Lock的实现者需要实现的newCondition，就是用来产生这个对象的。具体的用法网上有很多。AQS的等待队列特点如下：

可以有多个等待队列；

同样提供

signal

和

signalAll

来通知一个或多个线程；

调用

await

相当于从同步队列中移出头节点，同时修改waitStatus并加入等待队列，也正因此，没有获取同步状态的线程，调用

await

将抛出异常；

调用

signal

相当于将当前等待队列的头结点移入同步队列；

由于等待队列中的节点不需要什么通信，所以一个单向链表就足以完成功能。实际上每个节点会直接自旋，所以不用前驱节点来唤醒，后继节点也就不用修改前驱节点的状态，比如前述的后继就需要动态修改前驱节点的状态为Node.SIGNAL；

与synchronized类似，进入等待队列的线程将会释放锁，这也是显然的，毕竟只有同步队列里的head节点才可能持有锁（这里针对于独占锁）；

支持超时等待、等待某个时间点以及响应中断等高级功能。

常见的使用方法是：

public class MyLock implement Lock {
private final AbstractQueuedSynchronizer sync = new Sync();
private class Sync extends AbstractQueuedSynchronizer {
// 实现同步器
}

···

public Condition newCondition() {
return sync.new ConditionObject();
}
}

由于ConditionObject内的

await

、

signal

等方法，都是在线程获取了锁（同步状态）后，才能调用，所以在ConditionObject内的实现比AQS本身要简单一些，不需要考虑循环CAS等确保线程安全的操作。

ConditionObject.await()

await

完成同步队列head结点到等待队列末尾的“转移”。实际上，这个过程如上述流程图所示，分两步完成的，首先构建节点加入等待队列队尾，然后head节点释放同步状态，最后看看这个节点的状态是否正常（可能因为各种原因，无法完成release）。整个过程都是在“锁”的保证之下的。看一个使用的实例：

public class TestCondition {
public static void main(String[] args) {
final ReentrantLock lock = new ReentrantLock();
final Condition ready = lock.newCondition();

Thread A = new Thread(new Runnable() {
public void run() {
lock.lock();
try {
ready.await();
System.out.println(Thread.currentThread().getName() + " says: I'm free");
} catch (Exception e) {

} finally {
lock.unlock();
}
}
}, "Waiter");

Thread B = new Thread(new Runnable() {
public void run() {
lock.lock();
try {
Thread.sleep(10000);
ready.signal();
System.out.println(Thread.currentThread().getName() + " says: send signal");

} catch (Exception e) {

} finally {
lock.unlock();
}
}
}, "Sender");

waiter.start();
sender.start();
}

一旦线程A调用

await()

并让出同步状态，只有当线程重新回到同步队列时，才能继续运行。否则，会借由

LockSupport.park

而进入WAIT状态。而线程A回到同步队列，是由另一个线程B调用

sign
c5fb
al()

来完成的。但是回到了同步队列后，线程可能依然保持WAIT状态，这是因为调用

signal()

还未释放锁，当B线程调用

unlock()

时才会重新使线程A变为RUNNING。调用

await

加入等待队列的流程图如下图所示。



下面是

await()

的源码，可以结合上图参阅：

public final void await() throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
// 会创建一个新的节点，加入等待队列
Node node = addConditionWaiter();
// 完全释放同步状态：也即认为当前线程占有着同步状态，不存在竞态
// 依次调用fullyRelease->release->tryRelease
// 如果释放失败，会直接抛出IllegalMonitorStateException
// 释放后，其他线程可以立即占有同步状态
int savedState = fullyRelease(node);
int interruptMode = 0;
while (!isOnSyncQueue(node)) {
LockSupport.park(this);
if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
break;
}
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
// 防止过早的signal，会直接跳过while循环，这时调用acquireQueued，会重新阻塞线程
// 这时park还未被调用，发送signal的线程也可能还未释放锁
// 如果不加以限制，await函数会直接返回，继续运行同步体中的内容
// 但其实线程并没有占有同步态
interruptMode = REINTERRUPT;
if (node.nextWaiter != null) // clean up if cancelled
unlinkCancelledWaiters();
if (interruptMode != 0)
reportInterruptAfterWait(interruptMode);
}

ConditionObject.signal()

承接上文内容，下面是线程B调用

singal

的流程图。



注意位于

transferForSignal

的重置前驱节点步骤，注释中给出的解释是，此时前驱节点失效，需要“resync”，于是重新启动了这个在

await

中等待的线程。设想一下，这个时候A线程实际上已经离开了等待队列，会调用

acquireQueue

（参看

await

的源码）。也就是说，完成整个节点从等待队列转移到同步队列过程的终态，必须和调用

acquire

获取同步状态相同：前驱节点状态为SIGNAL，线程自己调用LockSupport.park进入WAIT状态。附上源代码。

public final void signal() {
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
Node first = firstWaiter;
if (first != null)
doSignal(first);
}

private void doSignal(Node first) {
do {
if ( (firstWaiter = first.nextWaiter) == null)
lastWaiter = null;
first.nextWaiter = null;
} while (!transferForSignal(first) &&
(first = firstWaiter) != null);
}

final boolean transferForSignal(Node node) {
/*
* If cannot change waitStatus, the node has been cancelled.
*/
if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
return false;

/*
* Splice onto queue and try to set waitStatus of predecessor to
* indicate that thread is (probably) waiting. If cancelled or
* attempt to set waitStatus fails, wake up to resync (in which
* case the waitStatus can be transiently and harmlessly wrong).
*/
Node p = enq(node);
int ws = p.waitStatus;
// 重置前驱节点状态为SIGNAL
if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
LockSupport.unpark(node.thread);
return true;
}

参考文献

https://my.oschina.net/andylucc/blog/651982

《Java并发编程的艺术》 ↩
《Java并发编程的艺术》 ↩