深入理解Java并发机制（4）--AQS、ReentrantLock、ReentrantReadWriteLock源码分析

背景

锁是用来控制多个线程访问共享资源的工具，Java中有两种锁，一种是jvm层面的synchronized关键字，一种是JUC的Lock接口的实现类。

Lock接口对比synchronized关键字需要显示的获取/释放锁，虽然丧失了一定的便捷性，但是提供了synchronized关键字所不具备的获取/释放锁的灵活性、可中断获取锁、超时获取锁、非阻塞获取锁（trylock）、以及多个条件变量。

队列同步器（AQS），是用来构建锁和其他同步组件的基础框架，它提供了自动管理同步状态、线程的阻塞/唤醒、排队、条件变量以及监控五个方面的通用结构。AQS的设计模式基于模板方法模式，主要使用方式为继承，子类通过继承AQS并实现它的抽象方法来管理同步状态，相当于子类同步器将同步的管理委托给AQS。

本文将沿着锁的获取到释放这个过程深入到AQS的源码实现。

进一步认识锁

公平/非公平。

锁按照获取策略可以分为公平锁和非公平锁。公平锁即要获取锁的线程都要按先来后到的顺序排队。非公平锁即在条件允许的情况下，谁竞争成功则谁获取到锁，获取失败的线程排队。

共享/排他

按照对于共享资源的访问模式，锁可以分为共享锁和排他锁。当多个线程都是读操作时，对共享资源内容不会修改，可以安全的并发访问，使用共享锁可以提高系统吞吐量。当多个线程中含有写操作时，会修改共享资源的内容，必须要采用排他模式保证共享资源的线程安全。

可重入锁

也叫递归锁。已经获取锁的线程在执行过程中可以不用竞争直接继续获取锁，只需在同步状态中记录索取的次数即可。可重入锁提高效率的同时，也可以有效避免死锁的发生。

由之前的文章我们可知synchronized关键字提供的是非公平的、排他、可重入的锁。由AQS框架支撑的可重入锁ReentrantLock可以选择公平性，ReentrantReadWriteLock可以选择公平性、共享性。这里就可以体现出Lock接口下的锁的灵活之处。

这里有个线程饥饿的问题，就是当获取策略为非公平，共享锁和排他锁共存的情况，有可能造成排他锁线程长期饥饿状态得不到运行的情况。需要在排队时做出一定的限制，保证排他锁也有机会获取锁。

初识AQS

AQS主要提供了同步状态的管理、线程的阻塞/唤醒、排队、条件变量、监控五个方面的功能。

同步状态的管理

AQS有一个volatile修饰的int变量state用于表示同步状态，在前面的文章我们讲解过volatile关键字的作用。同时提供了三个子类可访问的getState()，setState(int newState)，compareAndSetState(int expect, int update)的方法来访问和修改同步状态。

线程获取同步锁的策略由抽象方法tryAcquire(int arg)，tryRelease(int arg)，tryAcquireShared(int arg)，tryReleaseShared(int arg)描述，交由子类自己实现。这里AQS将操作和策略解耦，AQS本身并不关心子类怎么定义获取，怎么定义获取成功与失败，AQS关心的是当线程获取失败后如何处理（阻塞/唤醒、排队等），将获取策略解耦交给子类可以借以实现更复杂更灵活的同步方案（如排他模式下的公平锁和非公平锁ReentrantLock，共享模式下的ReentrantReadWriteLock等。）

排队

AQS内部维护了一个的带头尾引用的双链表用以构成一个FIFO的类CLH队列，队列中的节点类型由内部类Node表示。Node节点含有等待线程的引用、Node是共享/排他型节点、Node的状态、前后Node的引用等信息。head节点中的线程为当前队列中获取到锁的线程。

当线程竞争共享资源失败后，AQS将用该线程构建Node，并将其入队。

阻塞与唤醒

线程在获取资源失败后在AQS中入队，该线程即将阻塞以释放CPU，这里Node节点的状态信息将发挥关键作用。

waitStatus 节点状态	意义
CANCELLED	值为1，在线程在排队过程中出现了超时或者中断时，该线程将被取消，此时该节点状态将被置为1
SIGNAL	值为-1，由后继节点设置，当后继节点将当前节点状态设置为-1后，后继节点进入阻塞态；当前节点如果释放了资源或者被取消，将会通知后继节点（唤醒它）
CONDITION	值为-2，当节点在条件变量上等待时，节点的状态为CONDITION
PROPAGATE	值为-3，用于共享模式，表示下一次共享式同步状态将会无条件地被传播下去
INITIAL	值为0，初始状态

线程入队后，检查前驱节点是否被取消，如果被取消则向前进直到找到未取消的前驱节点，将该前驱节点的状态设置为SIGNAL，然后进入阻塞态。

head节点为队列中当前占用资源的线程，该线程释放资源时，将唤醒后继节点竞争锁。

这类似等待/通知模式，这种方式避免了自旋主动查询浪费CPU时间。

条件变量

条件变量常与互斥量一起使用，在条件变量上的等待需要先获取互斥量。AQS的等待/通知机制由内部类ConditionObject支持。ConditionObject实现了Condition接口，内部维护了一个带有首尾引用的FIFO单链表。节点类型依然为Node。

当有线程在某个ConditionObject上等待（调用Condition.await()）时，那么该线程首先将释放锁，然后构造新的Node类型节点入队并进入阻塞状态。

调用ConditionObject对象上的signal()方法，先将首节点移至同步等待队列队尾，然后唤醒该线程，进而加入到同步状态的竞争中。

监控

AQS提供了一些用于监控当前同步队列和条件队列的方法。如getQueueLength()，getQueuedThreads()，getExclusiveQueuedThreads()，getSharedQueuedThreads()，getWaitQueueLength(ConditionObject condition)等，便于对同步器的竞争状态有个估计（AQS中的队列一直在变化，监测方法只能返回一个估计值）。

锁有两个基本的操作，一个是锁的获取，一个是锁的释放。

AQS支持不可中断、可中断、超时三种类型的获取锁，后两者均基于不可中断的获取模式；同时AQS还支持共享锁和排他锁，这里我们先以不可中断方式、排他模式进行分析。

排它锁

获取过程

AQS的acquire方法：

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

代码逻辑：先尝试获取（此方法由子类自己定义），如果尝试失败了，则新增节点（addWaiter），再进行排队（acquireQueued）。新增节点是先将当前线程构建新节点，然后CAS设置tail节点入队。重点在入队之后的acquireQueued方法：

final boolean acquireQueued(final Node node, int
4000
arg) {
boolean failed = true;
try {
boolean interrupted = false;
for (;;) {
//进入队列后进入循环
final Node p = node.predecessor();
//当前驱节点为head节点时，当前节点具有竞争资格，如果竞争成功，则将当前节点设置为头结点，并从循环中退出。也就是说，队列中只有头结点是占有锁的。
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null; // help GC
failed = false;
return interrupted;
}
//如果前驱节点不是head节点，或者竞争失败，看下面的shouldParkAfterFailedAcquire(p, node)方法
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus;
//如果前驱节点状态是SIGNAL，则表示前驱节点在释放资源时会通知自己，这时候返回true，由parkAndCheckInterrupt()方法阻塞线程。
if (ws == Node.SIGNAL)
/*
* This node has already set status asking a release
* to signal it, so it can safely park.
*/
return true;
//如果前驱节点状态大于0，只可能是节点已经被取消，则不断前进直到找到未取消的节点
if (ws > 0) {
/*
* Predecessor was cancelled. Skip over predecessors and
* indicate retry.
*/
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else {
//除去上面两种状态，现在只可能是0或者PROPAGATE状态了，因为shouldParkAfterFailedAcquire是获取失败或者没资格获取后调用的方法，我们需要将前驱节点的状态CAS设置为SIGNAL，以便当资源可用时通知自己去竞争。
/*
* 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.
*/
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}

这里我们可以看到每个新节点入队后，都会在前驱节点的SIGNAL状态上阻塞。另外在当自己有了竞争资格但是竞争失败的情况下，如果头结点状态不为SIGNAL，自己可能还有几次的循环请求的机会。

至此，不可中断、排他模式的锁获取过程就完成了，获取失败的节点在队列中阻塞，等待前驱节点的唤醒。

释放过程

继续看排他模式，源码如下：

public final boolean release(int arg) {
if (tryRelease(arg)) {
Node h = head;
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}

代码逻辑：如果尝试释放锁成功，则调用unparkSuccessor方法唤醒头结点的后继节点。

private void unparkSuccessor(Node node) {
/*
* If status is negative (i.e., possibly needing signal) try
* to clear in anticipation of signalling.  It is OK if this
* fails or if status is changed by waiting thread.
*/
int ws = node.waitStatus;
if (ws < 0)
compareAndSetWaitStatus(node, ws, 0);

/*
* Thread to unpark is held in successor, which is normally
* just the next node.  But if cancelled or apparently null,
* traverse backwards from tail to find the actual
* non-cancelled successor.
*/
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)
s = t;
}
if (s != null)
LockSupport.unpark(s.thread);
}

排它锁实例ReentrantLock

ReentrantLock实现了Lock接口，提供公平与非公平两种模式，分别由静态内部类NonfairSync和FairSync支撑，这两个静态内部类都是抽象内部类Sync的子类，而Sync继承自AQS。

公平模式

可以看到公平模式的lock方法直接调用了AQS acquire()方法：

final void lock() {
acquire(1);
}

公平模式的tryAcquire方法：

protected final boolean tryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
if (c == 0) {
//在CAS设置同步状态之前，加了一个hasQueuedPredecessors()，如果已经有前驱在排队了，那么获取失败，也就是说对于每个线程都是严格排队的，即所谓公平。
if (!hasQueuedPredecessors() &&
compareAndSetState(0, acquires)) {
setExclusiveOwnerThread(current);
return true;
}
}
else if (current == getExclusiveOwnerThread()) {
int nextc = c + acquires;
if (nextc < 0)
throw new Error("Maximum lock count exceeded");
setState(nextc);
return true;
}
return false;
}

非公平模式

直接上代码：

final void lock() {
if (compareAndSetState(0, 1))
setExclusiveOwnerThread(Thread.currentThread());
else
acquire(1);
}

不管是否有前驱节点在排队，先直接以0为预期值CAS设置同步状态，如果成功了就设置当前线程为获取锁的线程。如果失败则调用AQS acquire获取锁逻辑（上面讲过，先尝试，不成功再排队）。并不考虑当前是否已经有前驱节点在排队，即所谓非公平。

我们再看一下NonfairSync所对应的tryAcquire方法：

final boolean nonfairTryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
if (c == 0) {
if (compareAndSetState(0, acquires)) {
setExclusiveOwnerThread(current);
return true;
}
}
//如果当前线程和当前拥有锁的线程是同一个线程，允许重入，只需要将同步状态+1表示重入次数即可。
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;
}

非公平模式相较于公平模式少了一些排队的开销，在实际应用中有更大的吞吐量。

公平模式和非公平模式的释放过程是一致的，都是直接调用AQS release(int arg)方法，我们来看其对应的tryRelease()方法：

protected final boolean tryRelease(int releases) {
//获得一次锁，同步状态+1，释放时每释放一次同步状态-1，直到状态为0，可重入锁全部释放完毕。
int c = getState() - releases;
if (Thread.currentThread() != getExclusiveOwnerThread())
throw new IllegalMonitorStateException();
boolean free = false;
if (c == 0) {
free = true;
setExclusiveOwnerThread(null);
}
setState(c);
return free;
}

如前文所述，AQS还支持共享模式获取锁。

共享锁

释放过程

因为共享锁的获取过程中调用了释放过程，我们这里先说明共享锁的释放过程。AQS中releaseShared()方法：

public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}

private void doReleaseShared() {
/*
* Ensure that a release propagates, even if there are other
* in-progress acquires/releases.  This proceeds in the usual
* way of trying to unparkSuccessor of head if it needs
* signal. But if it does not, status is set to PROPAGATE to
* ensure that upon release, propagation continues.
* Additionally, we must loop in case a new node is added
* while we are doing this. Also, unlike other uses of
* unparkSuccessor, we need to know if CAS to reset status
* fails, if so rechecking.
*/
for (;;) {
Node h = head;
if (h != null && h != tail) {
int ws = h.waitStatus;
if (ws == Node.SIGNAL) {
//暂时不清楚这里两个循环CAS的意义具体是啥
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
}
//这里也不是很清楚，如果一个共享节点的后续节点是排他节点，那么共享节点setHeadAndPropagate时，并不会唤醒后继节点。如果共享节点的后继是共享型的，那么该后继改变了head节点意味着它已经获取了锁，这里如果重新循环，那么不管该后继的后继是共享型的还是排他型的，该后继都会唤醒它的后继？
if (h == head)                   // loop if head changed
break;
}
}

获取过程

AQS的acquireShared方法：

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

先尝试获取共享锁，失败后进入doAcquireShared()方法，其中tryAcquireShared()方法由子类实现。

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) {
int r = tryAcquireShared(arg);
//tryAcquireShared返回int整数，负数表示失败，0表示成功但是后面不能再继续有线程获取共享锁了，正数表示成功并且后面的线程可以继续获取共享锁。
if (r >= 0) {
//如果获取成功，将当前节点设置为头节点并传播
setHeadAndPropagate(node, r);
p.next = null; // help GC
if (interrupted)
selfInterrupt();
failed = false;
return;
}
}
//获取失败则排队阻塞
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}

private void setHeadAndPropagate(Node node, int propagate) {
Node h = head; // Record old head for check below
setHead(node);
/*
* Try to signal next queued node if:
*   Propagation was indicated by caller,
*     or was recorded (as h.waitStatus) by a previous operation
*     (note: this uses sign-check of waitStatus because
*      PROPAGATE status may transition to SIGNAL.)
* and
*   The next node is waiting in shared mode,
*     or we don't know, because it appears null
*
* The conservatism in both of these checks may cause
* unnecessary wake-ups, but only when there are multiple
* racing acquires/releases, so most need signals now or soon
* anyway.
*/
//注意到在获取共享锁的时候，设置当前节点为头结点，如果允许其他线程继续获取共享锁（tryAcquireShared返回大于0），且下一个等待线程是共享类型的节点，将进入doReleaseShared()方法
if (propagate > 0 || h == null || h.waitStatus < 0) {
Node s = node.next;
if (s == null || s.isShared())
doReleaseShared();
}
}

共享锁在获取成功后，如果条件允许且下一个等待线程是共享类型的节点，那么该线程会唤醒它。与排他锁相比，共享锁不仅在释放锁时唤醒后继节点，在获取锁成功后亦会唤醒后继的共享节点。

共享锁/排他锁示例ReentrantReadWriteLock

ReentrantReadWriteLock实现了ReadWriteLock接口，同时提供读锁和写锁。它共有5个内部类：ReadLock，WriteLock，Sync，FairSync，NonfairSync。

Sync继承自AQS，不像ReentrantLock中的Sync，只实现了AQS的排他锁的获取与释放。这里一并实现了AQS的排它锁获取与释放（tryAcquire/tryRelease）和共享锁的获取和释放（tryAcquireShared/tryReleaseShared），AQS的灵活性可见一斑。

FairSync与NonfairSync继承Sync提供了公平性与非公平性的选择。

ReadLock与WriteLock均实现了Lock接口，两者使用同一个已确定公平性策略的Sync对象作为同步器来管理读锁和写锁的并发控制。

ReentrantReadWriteLock相比于ReentrantLock，多了共享锁，在读多于写的场景下，前者比后者能提供更大的吞吐量。

ReentrantReadWriteLock实现基于：并发读共享资源不会有线程安全问题，而只要操作中含有并发写就会带来线程安全问题，必须正确同步。并发读由共享锁-读锁支持，并发写由排他锁-写锁支持。

来看一下ReentrantReadWriteLock中的Sync是如何做到的吧。

同步状态的表示

由于AQS只有一个int型state变量表示同步状态，而读写锁需要同时表示读锁和写锁的状态，Sync中将该state变量一拆为二，高16位用于表示读锁状态，低16位用于表示写锁状态。

static int sharedCount(int c)    { return c >>> SHARED_SHIFT; }
static int exclusiveCount(int c) { return c & EXCLUSIVE_MASK; }

两个方法通过位运算在同一个int变量上维护了读锁和写锁的状态。

对于读锁，sharedCount返回值反映了当前获取读锁的线程数目，对于读线程的重入计数，由ThreadLocal的子类ThreadLocalHoldCounter变量readHolds支撑，该变量为每个读线程记录了重入数目，原理可以阅读ThreadLocal源码。

对于写锁，exclusiveCount返回值反应了当前写锁的获取状态和重入次数。

公平性策略

abstract boolean readerShouldBlock();
abstract boolean writerShouldBlock();

Sync定义了两个抽象方法，用来控制并发线程入队的公平性。

非公平：

static final class NonfairSync extends Sync {
private static final long serialVersionUID = -8159625535654395037L;
//非公平策略下，写线程总是可以直接竞争锁
final boolean writerShouldBlock() {
return false; // writers can always barge
}
//读线程在同步等待队列的第一个节点不为排他型节点时也可以直接竞争锁，但是如果第一个节点为排他型节点，该读线程将阻塞，这样可以防止在排队的写线程长期饥饿，不公平中为写线程争取了公平。
final boolean readerShouldBlock() {
/* As a heuristic to avoid indefinite writer starvation,
* block if the thread that momentarily appears to be head
* of queue, if one exists, is a waiting writer.  This is
* only a probabilis
1019e
tic effect since a new reader will not
* block if there is a waiting writer behind other enabled
* readers that have not yet drained from the queue.
*/
return apparentlyFirstQueuedIsExclusive();
}
}
//检查同步等待队列中的第一个节点是否是排他型，如果是排他型则返回true。
final boolean apparentlyFirstQueuedIsExclusive() {
Node h, s;
return (h = head) != null &&
(s = h.next)  != null &&
!s.isShared()         &&
s.thread != null;
}

公平：如果有前驱节点在排队，则阻塞。

static final class FairSync extends Sync {
private static final long serialVersionUID = -2274990926593161451L;
final boolean writerShouldBlock() {
return hasQueuedPredecessors();
}
final boolean readerShouldBlock() {
return hasQueuedPredecessors();
}
}

写锁的获取和释放

获取过程

获取写锁时必须没有读锁。

protected final boolean tryAcquire(int acquires) {
/*
* Walkthrough:
* 1. If read count nonzero or write count nonzero
*    and owner is a different thread, fail.
* 2. If count would saturate, fail. (This can only
*    happen if count is already nonzero.)
* 3. Otherwise, this thread is eligible for lock if
*    it is either a reentrant acquire or
*    queue policy allows it. If so, update state
*    and set owner.
*/
Thread current = Thread.currentThread();
int c = getState();
int w = exclusiveCount(c);
if (c != 0) {
// (Note: if c != 0 and w == 0 then shared count != 0)
//当state不为0，可能是有读锁或者是有写锁，如果写锁为0，当前线程不是拥有锁的线程，则获取失败。
if (w == 0 || current != getExclusiveOwnerThread())
return false;
if (w + exclusiveCount(acquires) > MAX_COUNT)
throw new Error("Maximum lock count exceeded");
// Reentrant acquire
//如果写锁不为0，那么读锁肯定为0，当前线程拥有锁，直接重入，获取成功。
setState(c + acquires);
return true;
}
//由公平性策略检查是否需要阻塞
if (writerShouldBlock() ||
!compareAndSetState(c, c + acquires))
return false;
setExclusiveOwnerThread(current);
return true;
}

释放过程

先检查当前线程是否拥有锁，然后将state写锁计数部分减去释放数。

protected final boolean tryRelease(int releases) {
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
int nextc = getState() - releases;
boolean free = exclusiveCount(nextc) == 0;
if (free)
setExclusiveOwnerThread(null);
setState(nextc);
return free;
}

读锁的获取与释放

获取过程

tryAcquireShared与fullTryAcquireShared代码有部分重复，tryAcquireShared就是先尝试获取，失败再进入fullTryAcquireShared，这里直接分析fullTryAcquireShared方法。读锁获取时必须没有写锁。

final int fullTryAcquireShared(Thread current) {
/*
* This code is in part redundant with that in
* tryAcquireShared but is simpler overall by not
* complicating tryAcquireShared with interactions between
* retries and lazily reading hold counts.
*/
HoldCounter rh = null;
for (;;) {
int c = getState();
if (exclusiveCount(c) != 0) {
if (getExclusiveOwnerThread() != current)
return -1;
// else we hold the exclusive lock; blocking here
// would cause deadlock.
//存在一个时刻，既没有读锁也没有写锁，但是此时可能一个读锁和一个同步队列中的写锁竞争
} else if (readerShouldBlock()) {
// Make sure we're not acquiring read lock reentrantly
if (firstReader == current) {
// assert firstReaderHoldCount > 0;
} else {
if (rh == null) {
rh = cachedHoldCounter;
if (rh == null || rh.tid != current.getId()) {
rh = readHolds.get();
if (rh.count == 0)
//帮助GC
readHolds.remove();
}
}
if (rh.count == 0)
return -1;
}
}
if (sharedCount(c) == MAX_COUNT)
throw new Error("Maximum lock count exceeded");
//将整个过程放在一个自旋里，就是有可能前面的条件允许，但是多个读线程同时CAS修改state值时有可能失败，自旋保证条件允许的情况下一定能线程安全的修改成功。
if (compareAndSetState(c, c + SHARED_UNIT)) {
//为相应的读线程记录重入数目，并且为最后一个访问的线程设置缓存
if (sharedCount(c) == 0) {
firstReader = current;
firstReaderHoldCount = 1;
} else if (firstReader == current) {
firstReaderHoldCount++;
} else {
if (rh == null)
rh = cachedHoldCounter;
if (rh == null || rh.tid != current.getId())
rh = readHolds.get();
else if (rh.count == 0)
readHolds.set(rh);
rh.count++;
cachedHoldCounter = rh; // cache for release
}
return 1;
}
}
}

释放过程

代码逻辑：先将ThreadLocal变量中对应线程的重入计数值减1，并注意清除不再使用的ThreadLocal变量帮助GC，最后用循环CAS设置state读数目。

protected final boolean tryReleaseShared(int unused) {
Thread current = Thread.currentThread();
if (firstReader == current) {
// assert firstReaderHoldCount > 0;
if (firstReaderHoldCount == 1)
firstReader = null;
else
firstReaderHoldCount--;
} else {
HoldCounter rh = cachedHoldCounter;
if (rh == null || rh.tid != current.getId())
rh = readHolds.get();
int count = rh.count;
if (count <= 1) {
readHolds.remove();
if (count <= 0)
throw unmatchedUnlockException();
}
--rh.count;
}
for (;;) {
int c = getState();
int nextc = c - SHARED_UNIT;
if (compareAndSetState(c, nextc))
// Releasing the read lock has no effect on readers,
// but it may allow waiting writers to proceed if
// both read and write locks are now free.
return nextc == 0;
}
}

条件变量

由于条件变量的条件常常与共享资源的状态有关，条件变量常常与互斥量一起使用，也就使得在条件变量上的wait()/notify()操作必须要先获取互斥量。ConditionObject作为AQS的内部类也合情合理。实际上，在Lock接口中的newCondition()方法也表明了ConditionObject对象对锁的依赖性。由此，对于ConditionObject类的方法就不需要再另外做同步了。

每个对象都有wait()/notify()方法，在synchronized关键字的实现中，每一个对象一个管程，获取了对象的管程之后，才能进行wait()/notify()操作，一个对象一个条件变量。在许多场景下，线程需要在多个条件变量上阻塞，这时候synchronized关键字就无能为力了。但是AQS的条件变量没有这样的限制，可以创建多个条件变量ConditionObject。

ConditionObject实现了Condition接口，内部维护了一个带有首尾引用的FIFO单链表。节点类型依然为Node。主要有三类方法：await、notify和监控方法，其中await也提供了不可中断、可中断、超时三类方法。

await/notify机制主要逻辑：

假设有线程A和线程B，线程A先获取锁，执行….，然后在某个conditionObject上await，调用await意味着线程A释放了其获得的锁并进入等待队列阻塞自己（注意：此时await方法并没有返回）。

线程B此时可以获取锁，执行….，在A线程等待的conditionObject上调用signal方法，将A线程从等待队列中移至锁的同步队列中去竞争锁，然后B线程释放锁。

此时A线程在同步队列中再次竞争到锁，然后此时await方法才返回，A线程继续执行await方法之后的代码。await有一个隐式的释放锁-阻塞-获取锁的过程。

await方法

当一个线程在条件对象上等待时，先检查条件队列中是否有已取消的线程，如果有则清除（这里由于条件队列是单链表，每次清除都需要从头到尾遍历一遍），然后以当前线程构建新的节点，节点类型为CONDITION，加入条件队列。

private Node addConditionWaiter() {
Node t = lastWaiter;
// If lastWaiter is cancelled, clean out.
if (t != null && t.waitStatus != Node.CONDITION) {
unlinkCancelledWaiters();
t = lastWaiter;
}
Node node = new Node(Thread.currentThread(), Node.CONDITION);
if (t == null)
firstWaiter = node;
else
t.nextWaiter = node;
lastWaiter = node;
return node;
}

public final void awaitUninterruptibly() {
Node node = addConditionWaiter();
//入队以后，当前线程需要释放锁，不论是读锁还是写锁，释放锁调用的是AQS的release方法，tryRelease后，就会唤醒同步队列中的后继节点竞争锁。
int savedState = fullyRelease(node);
boolean interrupted = false;
//然后线程循环检测自己是否在同步队列上（后面将分析signal()方法主要是将Condition节点移至锁对象所维护的同步队列上），如果不在同步队列表示自己仍需在条件对象上等待，如果已经在同步队列，线程从循环中退出。
while (!isOnSyncQueue(node)) {
LockSupport.park(this);
if (Thread.interrupted())
interrupted = true;
}
//等待的线程节点由等待队列进入同步队列，通过acquireQueued参与到锁的竞争中去。前面讲过这个方法，如果有机会就尝试获取锁。如果自己前面还有线程在等待，则将前者的状态设置为SIGNAL，然后自己阻塞等待前驱节点的唤醒。
//如果获取锁成功，再处理中断情况，await方法返回（条件已经满足了），继续执行后面的代码。也就是说await方法返回时，是一定再次获取了锁的。
if (acquireQueued(node, savedState) || interrupted)
selfInterrupt();
}

signal方法

siganl方法只做一个事情：将条件变量等待队列的节点移至锁的同步队列，这样在条件变量上等待的线程就可以参与锁的竞争，当那些线程获取到锁后即从await方法中返回，继续执行。signalAll即将所有等待队列上的节点都转移至同步队列参与竞争。

signal方法主要调用了transferForSignal方法：

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;
}

可中断和超时

AQS中获取锁的中断和ConditionObject的中断机制稍有不同，我们分开讲。

锁的获取

主要涉及的方法有acquire和acquireInterruptibly，acquireShared和acquireSharedInterruptibly。

再来看acquire和acquireInterruptibly的源码：

public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}public final void acquireInterruptibly(int arg)
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
if (!tryAcquire(arg))
doAcquireInterruptibly(arg);
}

主要区别在于tryAcquire失败后的处理，再看acquireQueued：

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;
//在成功获取锁后将中断信息返回，最终在acquire中再次调用selfInterrupt方法设置线程中断标志位
return interrupted;
}
if (shouldParkAfterFailedAcquire(p, node) &&
//在检查到被中断后只是将中断标志位设置为true
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}

而acquireInterruptibly呢：

private void doAcquireInterruptibly(int arg)
throws InterruptedException {
final Node node = addWaiter(Node.EXCLUSIVE);
boolean failed = true;
try {
for (;;) {
final Node p = node.predecessor();
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null; // help GC
failed = false;
return;
}
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
//在检查到有中断发生后直接抛异常
throw new InterruptedException();
}
} finally {
//抛异常会导致获取锁失败，进而该获取锁线程节点被取消
if (failed)
cancelAcquire(node);
}
}

acquireShared和acquireSharedInterruptibly的中断机制与上述一致，不再赘述。

ConditionObject–await方法的中断机制

在Doug Lea的论文The java.util.concurrent Synchronizer Framework中有这么一段话：

As revised in JSR133, these require that if an interrupt occurs before a signal,then the await method must, after re-acquiring the lock, throw InterruptedException. But if it is interrupted after a signal, then the method must return without throwing an exception, but with its thread interrupt status set.

按照JSR133的要求，在signal之前被中断的线程在重新获取锁后要抛InterruptedException，而在signal之后被中断的线程从await返回时不能抛异常，而是设置线程的中断标志位。也就是说在线程处于等待阶段时，抛出异常；在竞争锁阶段，设置标志位。

源码中也定义了两个变量来表示这两种情况：

/** Mode meaning to reinterrupt on exit from wait */
private static final int REINTERRUPT =  1;
/** Mode meaning to throw InterruptedException on exit from wait */
private static final int THROW_IE    = -1;

并在可中断的await方法后面都加入了方法：

private void reportInterruptAfterWait(int interruptMode)
throws InterruptedException {
if (interruptMode == THROW_IE)
throw new InterruptedException();
else if (interruptMode == REINTERRUPT)
selfInterrupt();
}

根据情况的不同await方法退出时有不同的处理。

以await()方法为例，我们来看它是如何做到的：

public final void await() throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
Node node = addConditionWaiter();
int savedState = fullyRelease(node);
int interruptMode = 0;

while (!isOnSyncQueue(node)) {
LockSupport.park(this);
//这里有一个check方法
if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
break;
}
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
interruptMode = REINTERRUPT;
if (node.nextWaiter != null) // clean up if cancelled
unlinkCancelledWaiters();
if (interruptMode != 0)
reportInterruptAfterWait(interruptMode);
}

解决问题的关键在于Node节点的状态。当节点处于等待队列中时，Node的状态为Condition。当节点被移至同步队列中后（signal后），Node的状态为0。

当线程在等待队列中等待时，如果有中断发生，线程从park()方法返回（park()方法也可能无端返回，这里略去），进入checkInterruptWhileWaiting方法：

private int checkInterruptWhileWaiting(Node node) {
return Thread.interrupted() ?
(transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) :
0;
}

重点看其中的transferAfterCancelledWait方法：

final boolean transferAfterCancelledWait(Node node) {
//如果此时CAS操作成功了，证明还没有signal信号将节点移至同步队列，也就是中断发生在等待阶段。那么本操作将节点移至同步队列，返回true。这里返回true，即指示await方法返回时要抛出异常
if (compareAndSetWaitStatus(node, Node.CONDITION, 0)) {
enq(node);
return true;
}
/*
* If we lost out to a signal(), then we can't proceed
* until it finishes its enq().  Cancelling during an
* incomplete transfer is both rare and transient, so just
* spin.
*/
//如果操作失败了，说明该节点要么已经在同步队列上了，要么就在去同步队列的路上，等待一下
while (!isOnSyncQueue(node))
Thread.yield();
return false;//返回false则指示await()返回时设置中断标志位 不抛异常
}

再来分析await方法中进入同步队列后的情况：

//如果在获取锁的过程中发生中断，属于设置标志位的情况。这里也可以看到调用的是acquire方法中的acquireQueued方法，不可中断，即中断发生了我只是做记录，并不对中断做什么处理，做下记录留给专门处理中断的代码去处理。
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
interruptMode = REINTERRUPT;
if (node.nextWaiter != null) // clean up if cancelled
unlinkCancelledWaiters();
if (interruptMode != 0)
//如果这两个阶段发生了中断的话，最后由该方法向调用线程报告中断情况，是抛出异常还是仅仅设置标志位
reportInterruptAfterWait(interruptMode);
}

超时

超时部分比较简单，获取锁的过程中超时了直接返回获取失败，等待过程中超时了直接不再等待，移至同步队列中竞争锁。

监控方法

AQS中提供了不少监控同步队列和等待队列的状态的方法，这些方法在并发环境下获取的信息都是估计值，瞬时值。

Java并发编程的艺术

JSR133

Doug Lea的论文The java.util.concurrent Synchronizer Framework