上一篇介绍了AQS的基本设计思路以及两个内部类Node和ConditionObject的实现 聊聊高并发(二十一)解析java.util.concurrent各个组件(三) 深入理解AQS(一) 这篇说一说AQS的主要方法的实现。AQS和CLHLock的最大区别是,CLHLock是自旋锁,而AQS使用Unsafe的park操作让线程进入等待(阻塞)。
线程加入同步队列,和CLHLock一样,从队尾入队列,使用CAS+轮询的方式实现无锁化。入队列后设置节点的prev和next引用,形成双向链表的结构
-
private Node enq(final Node node) {
-
for (;;) {
-
Node t = tail;
-
if (t == null) { // Must initialize
-
if (compareAndSetHead(new Node()))
-
tail = head;
-
} else {
-
node.prev = t;
-
if (compareAndSetTail(t, node)) {
-
t.next = node;
-
return t;
-
}
-
}
-
}
-
}
线程指定独享还是共享方式加入队列,先尝试加入一次,如果失败再用enq()轮询地尝试,比如addWaiter(Node.EXCLUSIVE), addWaiter(Node.SHARED)
-
private Node addWaiter(Node mode) {
-
Node node = new Node(Thread.currentThread(), mode);
-
// Try the fast path of enq; backup to full enq on failure
-
Node pred = tail;
-
if (pred != null) {
-
node.prev = pred;
-
if (compareAndSetTail(pred, node)) {
-
pred.next = node;
-
return node;
-
}
-
}
-
enq(node);
-
return node;
-
}
唤醒后继节点,最典型的情况就是在线程释放锁后,会唤醒后继节点。会从节点的next开始,找到一个后继节点,如果next是null,就从队尾开始往head找,直到找到最靠近当前节点的后续节点。 waitStatus <= 0的隐含意思是线程没有被取消。 然后用LockSupport唤醒这个找到的后继节点的线程。
这个方法类似于CLHLock里面释放锁时,通知后续节点来获取锁。AQS使用了阻塞的方式,所以这个方法的后续方法是acquireXXX方法,它负责将后续节点唤醒,后续节点再根据状态去判断是否获得锁
-
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);
-
}
共享模式下的释放操作,从队首开始向队尾扩散,如果节点的waitStatu是SIGNAL,就唤醒后继节点,如果waitStatus是0,就设置标记成PROPAGATE
-
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;
-
}
-
}
取消获取操作,要把节点从同步队列中去除,通过链表操作将它的前置节点的next指向它的后继节点集合。如果该节点是在队尾,直接删除即可,否则要通知后继节点去获取锁
-
private void cancelAcquire(Node node) {
-
// Ignore if node doesn't exist
-
if (node == null)
-
return;
-
node.thread = null;
-
// Skip cancelled predecessors
-
Node pred = node.prev;
-
while (pred.waitStatus > 0)
-
node.prev = pred = pred.prev;
-
// predNext is the apparent node to unsplice. CASes below will
-
// fail if not, in which case, we lost race vs another cancel
-
// or signal, so no further action is necessary.
-
Node predNext = pred.next;
-
// Can use unconditional write instead of CAS here.
-
// After this atomic step, other Nodes can skip past us.
-
// Before, we are free of interference from other threads.
-
node.waitStatus = Node.CANCELLED;
-
// If we are the tail, remove ourselves.
-
if (node == tail && compareAndSetTail(node, pred)) {
-
compareAndSetNext(pred, predNext, null);
-
} else {
-
// If successor needs signal, try to set pred's next-link
-
// so it will get one. Otherwise wake it up to propagate.
-
int ws;
-
if (pred != head &&
-
((ws = pred.waitStatus) == Node.SIGNAL ||
-
(ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) &&
-
pred.thread != null) {
-
Node next = node.next;
-
if (next != null && next.waitStatus <= 0)
-
compareAndSetNext(pred, predNext, next);
-
} else {
-
unparkSuccessor(node);
-
}
-
node.next = node; // help GC
-
}
-
}
独占模式并且不可中断地获取队列锁的操作,这个方法在ConditionObject.await()中被使用,当线程被Unsafe.unpark唤醒后,需要调用acquireQueued来获取锁,从而结束await(). accquireQueued()方法要么获得锁,要么被tryAcquire方法抛出的异常打断,如果抛出异常,最后在finally里面取消获取
值得注意的是只有节点的前驱节点是head的时候,才能获得锁。这里隐含了一个意思,就是head指向当前获得锁的节点。当程序进入if(p == head and tryAcquire(arg))这个分支时,表示线程获得了锁或者被中断,将自己设置为head,将next设置为null.
shouldParkAfterFailedAcquired()方法的目的是将节点的前驱节点的waitStatus设置为SIGNAL,表示会通知后续节点,这样后续节点才能放心去park,而不用担心被丢失唤醒的通知。
parkAndCheckInterupt()方法会真正执行阻塞,并返回中断状态,这个方法有两种情况返回,一种是park被unpark唤醒,这时候中断状态为false。另一种情况是park被中断了,由于这个accquireQueued方法是不可中断的版本,所以即使线程被中断了,也只是设置了中断标志为true,没有跑出中断异常。在支持中断的获取版本里,这时会抛出中断异常。
这个方法可以理解为Lock的lock里没有获取锁的分支,在CLHLock自旋锁的实现里,是对前驱节点的状态自旋,而AQS是阻塞,所以这里是在同步队列里面进入了阻塞状态,等待被前驱节点释放锁时唤醒。
释放锁时会根据状态调用unparkSuccessor()方法来唤醒后续节点,这样就会在这个方法里面把阻塞的线程唤醒并获得锁。
队列锁的好处是线程都在多个共享状态上自旋或阻塞,所以unparkSuccessor()方法只会唤醒它后继没有取消的节点。
而取消只有两种情况,中断或者超时
-
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);
-
}
-
}
独占模式支持中断的获取队列锁操作,可以看到和不支持中断版本的区别,这里如果parkAndCheckInterrupt()方法返回时显示被中断了,就抛出中断异常
-
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);
-
}
-
}
独占模式限时获取队列锁操作, 这个获取的整体逻辑和前面的类似,区别是它支持限时操作,如果等待时间大于spinForTimeoutThreshold,就使用阻塞的方式等待,否则用自旋等待。使用了LockSupport.parkNanos()方法来实现限时地等待,并支持中断
这里隐含的一个含义是parkNanos方法退出有3种方式,
1. 限时到了自动退出,这时候会超时
2. 没有到限时被唤醒了,这时候是不超时的
3. 被中断
-
private boolean doAcquireNanos(int arg, long nanosTimeout)
-
throws InterruptedException {
-
long lastTime = System.nanoTime();
-
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 true;
-
}
-
if (nanosTimeout <= 0)
-
return false;
-
if (shouldParkAfterFailedAcquire(p, node) &&
-
nanosTimeout > spinForTimeoutThreshold)
-
LockSupport.parkNanos(this, nanosTimeout);
-
long now = System.nanoTime();
-
nanosTimeout -= now - lastTime;
-
lastTime = now;
-
if (Thread.interrupted())
-
throw new InterruptedException();
-
}
-
} finally {
-
if (failed)
-
cancelAcquire(node);
-
}
-
}
共享模式获得队列锁操作,获得操作也是从head的下一个节点开始,和独占模式只unparkSuccessor一个节点不同,共享模式下,等head的后续节点被唤醒了,它要扩散这种共享的获取,使用setHeadAndPropagate操作,把自己设置为head,并且把释放的状态往下传递,这里采用了链式唤醒的方法,1个节点负责唤醒1个后续节点,直到不能唤醒。当后继节点是共享模式isShared,就调用doReleaseShared来唤醒后继节点
doReleaseShared会从head开始往后检查状态,如果节点是SIGNAL状态,就唤醒它的后继节点。如果是0就标记为PROPAGATE, 等它释放锁的时候会再次唤醒后继节点。
这里有个隐含的意思:
1. 加入同步队列并阻塞的节点,它的前驱节点只会是SIGNAL,表示前驱节点释放锁时,后继节点会被唤醒。shouldParkAfterFailedAcquire()方法保证了这点,如果前驱节点不是SIGNAL,它会把它修改成SIGNAL。这里不是SIGNAL就有可能是PROPAGATE
2. 造成前驱节点是PROPAGATE的情况是前驱节点获得锁时,会唤醒一次后继节点,但这时候后继节点还没有加入到同步队列,所以暂时把节点状态设置为PROPAGATE,当后继节点加入同步队列后,会把PROPAGATE设置为SIGNAL,这样前驱节点释放锁时会再次doReleaseShared,这时候它的状态已经是SIGNAL了,就可以唤醒后续节点了
-
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);
-
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.
-
*/
-
if (propagate > 0 || h == null || h.waitStatus < 0) {
-
Node s = node.next;
-
if (s == null || s.isShared())
-
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;
-
}
-
}
tryXXXX 方法,这几个方法是给子类重写的,用来扩展响应的同步器操作
-
protected boolean tryAcquire(int arg) {
-
throw new UnsupportedOperationException();
-
}
-
protected boolean tryRelease(int arg) {
-
throw new UnsupportedOperationException();
-
}
-
protected int tryAcquireShared(int arg) {
-
throw new UnsupportedOperationException();
-
}
-
protected boolean tryReleaseShared(int arg) {
-
throw new UnsupportedOperationException();
-
}
独占模式获取操作的顶层方法,如果没有tryAcquired,或者没有获得队列锁,就中断
-
public final void acquire(int arg) {
-
if (!tryAcquire(arg) &&
-
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
-
selfInterrupt();
-
}
独占模式释放操作的顶层方法,如果tryRelease()成功,那么就唤醒后继节点去获取锁
-
public final boolean release(int arg) {
-
if (tryRelease(arg)) {
-
Node h = head;
-
if (h != null && h.waitStatus != 0)
-
unparkSuccessor(h);
-
return true;
-
}
-
return false;
-
}