AQS - 独占锁.md
JUC 系列之 AQS。
本文主要介绍 AQS ，其实现独占锁的方式与 Reentrantlock
AQS 全称 AbstractQueuedSynchronizer 也称为抽象队列同步器，是 java.util.concurrent.locks 包中的一个抽象类。JUC 中的许多锁比如 ReentrantLock 等都是基于 AQS 实现。其内部实现了一个等待队列，同时也使用了 ConditionObject 类以实现多条件唤醒。
现在来看看其源码；
AOS
根据图中可以看出，AQS 派生于 AOS，而 AOS 的源码比较简单，主要是一个 Thread 类型的私有变量以及 protected 的 getter/setter 方法。
Node
AQS 采用的是链队列的形式保存阻塞线程，其 静态内部类 Node 则为链表的结点，
首先是常量，分为 Node 类型，int 类型 与 VarHandle 类型，这里分别作说明：
- SHARED ：表示该 Node 为共享锁
- EXCLUSIVE ：表示该 Node 为独占锁
这两个常量主要赋值到 nextWaiter 变量中，其中 EXCLUSIVE 为 null，即 nextWaiter 为 null 表示该 Node 用于独占锁。
- CANCELLED：该线程已经关闭
- SIGNAL：后一个线程需要 unpark
- CONDITION：该线程在等待其 condition (条件队列)
- PROPAGATE： 标识其为强制传播 (共享锁中使用)
这四个变量重要用于 waitWaiter 变量中，用于标识该结点的状态。
- NEXT
- PREV
- THREAD
- WAITSTATUS
这四个为 varhandle 对象，分别对应四个结点的变量，为四个变量提供 cas 操作。
独占锁上锁
独占锁上锁主要有两个方法：
- tryAcquire
- acquire
// AbstractQueuedSynchronizer#tryAcquire
protected boolean tryAcquire(int arg) {
throw new UnsupportedOperationException();
}
可以看到这个方法直接抛出异常，实际上，这个方法是由子类重写的，之后分析 AQS 子类时会看到。
// AbstractQueuedSynchronizer#acquire
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
这个方法为获取一个独占锁的方法，许多子类的上锁方法最终都是调用到这个方法，比如 ReentrantLock#lock()，其中参数 arg 为自定义的任何参数，最终都会传入 tryAcquire 方法，由子类规定。该方法首先尝试调用 tryAcquire，当尝试上锁失败后，需要阻塞并放入阻塞队列，这里来到 acquireQueued 方法，这里先看看 addWaiter 方法：
/**
* Creates and enqueues node for current thread and given mode.
*
* @param mode Node.EXCLUSIVE for exclusive, Node.SHARED for shared
* @return the new node
*/
private Node addWaiter(Node mode) {
Node node = new Node(mode);
for (;;) {
Node oldTail = tail;
if (oldTail != null) {
node.setPrevRelaxed(oldTail);
if (compareAndSetTail(oldTail, node)) {
oldTail.next = node;
return node;
}
} else {
initializeSyncQueue();
}
}
}
这里的参数虽然是 Node 类型，不过这里表示的是结点的模式，也就是上面的 nextWaiter。
可以看到这里使用自旋的方式，先判断队尾是否为 Null，如果是 Null 则调用 initializeSyncQueue 方法初始化队列（为队列加上头节点），其中使用了 CAS 操作来更新头节点，而更新失败时不做任何处理。
如果当前队列以及初始化完毕，则先调用 node.setPrevRelaxed 将待入队的结点接上前一个结点，然后调用使用 CAS 操作将队尾结点 tail 更新为新的节点，成功时再将原本的尾结点接上新节点。
如果 cas 操作失败，则会重新进行 CAS 操作，直到成功。
然后返回新结点。
回到刚刚的方法，在调用 addWaiter 方法往队尾新增一个结点后，该节点会来到 acquireQueued 方法：
/**
* Acquires in exclusive uninterruptible mode for thread already in
* queue. Used by condition wait methods as well as acquire.
*
* @param node the node
* @param arg the acquire argument
* @return {@code true} if interrupted while waiting
*/
final boolean acquireQueued(final Node node, int arg) {
boolean interrupted = false;
try {
for (;;) {
final Node p = node.predecessor();
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null; // help GC
return interrupted;
}
if (shouldParkAfterFailedAcquire(p, node))
interrupted |= parkAndCheckInterrupt();
}
} catch (Throwable t) {
cancelAcquire(node);
if (interrupted)
selfInterrupt();
throw t;
}
}
该方法的作用我们可以称之为线程调度，即决定是否挂起该线程，主要决定几件事：
- 如果该节点是第一个节点（因为队列是有头节点的），则尝试加锁，如果成功了则直接出队（直接将该节点作为队列头结点），并返回标志。
- 如果不是，则调用
shouldParkAfterFailedAcquire来判断是否要将该节点挂起。在竞争不激烈的时候不挂起，让其保持自旋可以提升效率，同时，如果要将一个线程挂起，需要保证其前一个结点的 waitstatus 为 SIGNAL ，否则该线程挂起后可能不被唤醒，则可能卡死整个队列。
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus;
if (ws == Node.SIGNAL)
/*
* This node has already set status asking a release
* to signal it, so it can safely park.
*/
return true;
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 {
/*
* 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.
*/
pred.compareAndSetWaitStatus(ws, Node.SIGNAL);
}
return false;
}
可以看到根据 前一个结点的 SIGNAL 进行操作，当前一个结点的 waitStatus 为 SIGNAL 时，当它释放锁时会唤醒其下一个线程，因此这里可以放心挂起。如果 waitStatus > 0 则表示前一个结点已经被关闭了，则这里直接将该结点删除，一直到一个没有被关闭的结点。最后 将前一个结点的 waitStatus 手动设置为 SIGNAL 。
这里设置后没有直接返回 true，给了调用处几次自旋的机会，如果这一次还没获取锁，则会直接挂起。
当可以挂起时，会调用 parkAndCheckInterrupt() 将该线程挂起
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this);
return Thread.interrupted();
}
当挂起后，线程会在这暂停，也就是挂起的线程最终会停在 acquireQueued 方法，当某个地方将其唤醒后，它会继续该循环，重新尝试获取锁。
独占锁释放锁
释放独占锁非常 Easy，两个方法：
- tryRelease
- release
protected boolean tryRelease(int arg) {
throw new UnsupportedOperationException();
}
该方法与 tryAcquire 方法类似，AQS 中是直接抛异常，这里需要子类实现。
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 方法也比较简单，首先尝试释放锁，如果可以释放，则获取头节点，如果头节点不为 null 且头节点不是默认头节点 (只有初始的头节点 waitStatus 会为 0)
则调用 unparkSuccessor 方法唤醒 头节点的下一个结点 (这里 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)
node.compareAndSetWaitStatus(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 p = tail; p != node && p != null; p = p.prev)
if (p.waitStatus <= 0)
s = p;
}
if (s != null)
LockSupport.unpark(s.thread);
}
首先将结点的 waitStaus 设置为 0 默认模式，给之后的结点自旋机会。然后将后面所有被关闭的结点都删除找到第一个合法的结点，将其 unpark 。
注意，这里不进行任何队列的操作，因为将其 unpark 之后，线程会从 acquireQueued 继续执行，该方法会尝试获取锁等操作，释放锁只需要将其唤醒即可。
可中断独占锁
除了以上的 上锁方法， AQS 还提供了带超时时间的加锁方法，具体方法为 tryAcquireNanos 方法：
public final boolean tryAcquireNanos(int arg, long nanosTimeout) throws InterruptedException {
if (Thread.interrupted()) {
throw new InterruptedException();
} else {
return this.tryAcquire(arg) || this.doAcquireNanos(arg, nanosTimeout);
}
}
参数第一个为 tryAcquire 的参数，第二个为时间，这里以微秒为单位。
首先需要判断线程的中断标记，当中断标记为真的时候，是无法将线程挂起的（会抛异常），因此这里直接抛异常。
接下来会调用 tryAcquire 方法，当返回 true 时直接整个返回，否则调用 doAcquireNanos 方法：
private boolean doAcquireNanos(int arg, long nanosTimeout) throws InterruptedException {
if (nanosTimeout <= 0L) {
return false;
} else {
long deadline = System.nanoTime() + nanosTimeout;
AbstractQueuedSynchronizer.Node node = this.addWaiter(AbstractQueuedSynchronizer.Node.EXCLUSIVE);
try {
do {
AbstractQueuedSynchronizer.Node p = node.predecessor();
if (p == this.head && this.tryAcquire(arg)) {
this.setHead(node);
p.next = null;
return true;
}
nanosTimeout = deadline - System.nanoTime();
if (nanosTimeout <= 0L) {
this.cancelAcquire(node);
return false;
}
if (shouldParkAfterFailedAcquire(p, node) && nanosTimeout > 1000L) {
LockSupport.parkNanos(this, nanosTimeout);
}
} while(!Thread.interrupted());
throw new InterruptedException();
} catch (Throwable var8) {
this.cancelAcquire(node);
throw var8;
}
}
}
首先判断时间合法性，然后计算 死线。在生成结点。然后是一个自旋，如果当前队列第一个为该结点，并且 tryAcquire 返回 true，则直接出队返回 true。否则则判断当前时间，如果时间过了，则取消该结点 （调用 cancelAcquire 方法）。然后进行判断，如果需要挂起，并且当前时间差大于 1000L 微秒也就是 1 毫秒，则挂起，并且指定挂起时间为 超时时间。
ReentrantLock
ReentrantLock 是 使用 AQS 实现的可重入独占锁，同时支持公平和非公平。
首先该类不是直接继承于 AQS，而是持有了一个继承于 AQS 的内部类，先来看看类图：
可以看到，ReentrantLock 实现了 Lock 接口和 Serializable 接口。而 内部类 Sync 就继承于 AQS 类。同时 Sync 有两个子类分别对应公平锁和非公平锁的实例。先来看看 构造方法：
public ReentrantLock() {
this.sync = new ReentrantLock.NonfairSync();
}
public ReentrantLock(boolean fair) {
this.sync = (ReentrantLock.Sync)(fair ? new ReentrantLock.FairSync() : new ReentrantLock.NonfairSync());
}
构造 Sync 对象，根据参数判断实例。
然后是各种上锁和解锁方法：
public void lock() {
this.sync.acquire(1);
}
public boolean tryLock() {
return this.sync.nonfairTryAcquire(1);
}
public boolean tryLock(long timeout, TimeUnit unit) throws InterruptedException {
return this.sync.tryAcquireNanos(1, unit.toNanos(timeout));
}
public void unlock() {
this.sync.release(1);
}
可以看到 除了 tryLock 方法之外 主要是调用了 AQS 的相关方法。因此这里主要来到 Sync ：
主要看两个方法：
- nonfairTryAcquire
- tryRelease
第一个为非公平的尝试加锁，因为只有非公平锁才能尝试获取锁，因此 无论是公平锁还是非公平锁，当调用 tryLock 的时候必定会以非公平锁的形式去尝试获取锁，因此这里将此类写道 Sync 里，同时 tryLock 也是调用该方法：
@ReservedStackAccess
final boolean nonfairTryAcquire(int acquires) {
Thread current = Thread.currentThread();
int c = this.getState();
if (c == 0) {
if (this.compareAndSetState(0, acquires)) {
this.setExclusiveOwnerThread(current);
return true;
}
} else if (current == this.getExclusiveOwnerThread()) {
int nextc = c + acquires;
if (nextc < 0) {
throw new Error("Maximum lock count exceeded");
}
this.setState(nextc);
return true;
}
return false;
}
首先 Sync 中维护了一个变量 state，该变量是一个计数器，表示重入次数。因为对于可重入锁，需要记录重入次数，当次数为 0 时才可解锁。
首先获取一下当前的 state，如果为 0 则表示当前无锁，这里使用 CAS 操作尝试将 state 设置为 acquires，如果失败则直接返回 false，如果成功则将 exclusiveOwnerThread 赋值为当前线程 （因为 AQS 继承于 AOS），并返回 true。
如果当前 state 不为零，则表示目前处于加锁状态，则判断线程是否是当前线程，如果是的话直接将 state + acquires (如果溢出则抛异常)并返回 true。
最后返回 false。
该操作直接竞争锁，然后通过 CAS 操作获取锁，失败则直接返回 false，不会阻塞或自旋。
@ReservedStackAccess
protected final boolean tryRelease(int releases) {
int c = this.getState() - releases;
if (Thread.currentThread() != this.getExclusiveOwnerThread()) {
throw new IllegalMonitorStateException();
} else {
boolean free = false;
if (c == 0) {
free = true;
this.setExclusiveOwnerThread((Thread)null);
}
this.setState(c);
return free;
}
}
尝试释放锁，如果当前线程不持有锁则直接抛异常。将 state - releases，并判断是否为 0 ，如果为 0 则释放锁并返回 true。
这里不存在同步问题，因为对于独占锁来说，释放锁必须是当前线程，也就是说只有当前获取锁的线程才会调用该方法，其他线程要么在自旋等待锁，要么已经被挂起了，因此这里直接 set 即可，不需要 cas 操作。
然后来看看公平锁的对象：FairSync
继承于 Sync，重写了方法 tryAcquire，也就是 AQS 中需要重写的上锁方法
@ReservedStackAccess
protected final boolean tryAcquire(int acquires) {
Thread current = Thread.currentThread();
int c = this.getState();
if (c == 0) {
if (!this.hasQueuedPredecessors() && this.compareAndSetState(0, acquires)) {
this.setExclusiveOwnerThread(current);
return true;
}
} else if (current == this.getExclusiveOwnerThread()) {
int nextc = c + acquires;
if (nextc < 0) {
throw new Error("Maximum lock count exceeded");
}
this.setState(nextc);
return true;
}
return false;
}
}
公平锁的 上锁 是不能直接使用 CAS 操作的，所有锁都必须进入阻塞队列，因此与之前相比，多了一个 this.hasQueuedPredecessors() 的条件，该方法会返回当前线程之前是否有线程在排队，当队列不为空且队首不是当前线程时返回 false。当然该方法也会顺便清除无效的结点：
// AbstractQueuedSynchronizer#hasQueuedPredecessors()
public final boolean hasQueuedPredecessors() {
AbstractQueuedSynchronizer.Node h;
if ((h = this.head) != null) {
AbstractQueuedSynchronizer.Node s;
if ((s = h.next) == null || s.waitStatus > 0) {
s = null;
for(AbstractQueuedSynchronizer.Node p = this.tail; p != h && p != null; p = p.prev) {
if (p.waitStatus <= 0) {
s = p;
}
}
}
if (s != null && s.thread != Thread.currentThread()) {
return true;
}
}
return false;
}
之后时非公平锁：NonfairSync
static final class NonfairSync extends ReentrantLock.Sync {
private static final long serialVersionUID = 7316153563782823691L;
NonfairSync() {
}
protected final boolean tryAcquire(int acquires) {
return this.nonfairTryAcquire(acquires);
}
}
可以看到直接调用了父类的方法。
我们注意到它将 nonfairTyrAcquire 放到了 Sync 中，也就是说如果当前锁为公平锁对象，你依然可以调用非公平的尝试获取锁操作。这里主要是为了 tryLock 方法服务。对于 Lock 你需要实现其 tryLock 方法，而不管最终实现的什么锁，包括公平锁。