JDK1.5之后JAVA团队为开发者提供了一套面向并发编程的API,他们位于juc包(java.util.concurrent)中，其中包括了锁，同步容器类，支持原子操作类型的类，线程池等，而其中锁在开发中使用较多的工具之一，如ReentrantLock,CountDownLatch等，他们都是基于juc包中的java.util.concurrent.locks.AbstractQueuedSynchronizer类实现的。
本篇文章我们将通过源码的角度来分析AbstractQueuedSynchronizer的实现原理，分析下它本身有哪些能力，并给开发者提供了哪些扩展能力。
AbstractQueuedSynchronizer位于java.util.concurrent.locks包下面，
如下图所示：
AbstractQueuedSynchronizer：
它是一个提供实现阻塞锁和相关锁（同步器(信号量、事件等)）的基础框架，它依赖一个先进先出(FIFO)等待队列。它通过一个原子性的整形值表示状态，子类必须实现更改此状态的受保护方法，哪个对象调用该方法改变状态则说明哪个对象在获取或释放资源。这个类默认支持独占锁和共享锁两种模式。
既然它是一个框架，那么我就来了解下它是怎么构建框架的
1、首先它拥有一个队列，队列元素是内部内Node类的实例，下面我们看下Node的定义：
// 等待线程队列节点元素类，
<pre>
* +------+ prev +-----+ +-----+
* head | | <---- | | <---- | | tail
* +------+ +-----+ +-----+
* </pre>
先进先出的队列，一个线程入队只需要操作尾部节点（需要原子性），出队只需要更新头节点（不需要原子性）
static final class Node {
/** Marker to indicate a node is waiting in shared mode */
static final Node SHARED = new Node();
/** Marker to indicate a node is waiting in exclusive mode */
static final Node EXCLUSIVE = null;
/** 被取消状态，不再需要继续CAS获取锁 */
static final int CANCELLED = 1;
/** 后继线程需要停止休眠，继续运行起来 */
static final int SIGNAL = -1;
/** waitStatus value to indicate thread is waiting on condition */
static final int CONDITION = -2;
/**
* waitStatus value to indicate the next acquireShared should
* unconditionally propagate
*/
static final int PROPAGATE = -3;
/**
* Status field, taking on only the values:
* SIGNAL:
// 如果当前节点的后继节点是阻塞的，当前节点释放同步资源或者取消等待的时候必须唤醒它的后继节点，为了避免竞争，请求同步资源方法必须首先表明他们需要一个信号，重试原子获取同步资源，快速阻塞。
volatile int waitStatus;
/**
前驱结点
*/
volatile Node prev;
/**
* 后继节点
*/
volatile Node next;
/**
* 当前节点所属线程
*/
volatile Thread thread;
/**
* Link to next node waiting on condition, or the special
* value SHARED.
*
*/
Node nextWaiter;
/**
* Returns true if node is waiting in shared mode.
*/
final boolean isShared() {
return nextWaiter == SHARED;
}
/**
* Returns previous node, or throws NullPointerException if null.
* Use when predecessor cannot be null. The null check could
* be elided, but is present to help the VM.
*
* @return the predecessor of this node
*/
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;
}
}
// 线程节点入队列规则：
1、新节点前驱指向老尾节点
2、老尾节点后继指向新节点
3、尾节点设置为新的节点
/**
* Inserts node into queue, initializing if necessary. See picture above.
* @param node the node to insert
* @return node's predecessor
*/
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;
}
}
}
}
//先尝试当前线程快速原子性入等待队列操作，失败则进行原子性自旋的入队列操作
/**
* 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(Thread.currentThread(), mode);
// 先快速入等待队列，失败则进行自旋入队列，
Node pred = tail;
if (pred != null) {
node.prev = pred;
if (compareAndSetTail(pred, node)) {
pred.next = node;
return node;
}
}
enq(node);
return node;
}
// 设置头节点，并没有保障线程安全措施，达到出队列的操作，为了利于GC, 将头节点其他属性置空
/**
* Sets head of queue to be node, thus dequeuing. Called only by
* acquire methods. Also nulls out unused fields for sake of GC
* and to suppress unnecessary signals and traversals.
*
* @param node the node
*/
private void setHead(Node node) {
head = node;
node.thread = null;
node.prev = null;
}
//失败获取同步资源后，是否需要休眠，需要休眠的条件：
1、当前节点前驱节点是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与新的前驱节点绑定
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.
*/
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}
/** 唤醒后继节点
* Wakes up node's successor, if one exists.
*
* @param node the node
*/
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);
/*
// 当前线程释放锁后，唤醒后继线程所在的节点，如果后继几点为null或者节点为取消等待状态，则从队里尾部往头部遍历，直到找到一个可以继续获取同步资源的节点
* 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;
//如果当前节点后继节点为null,或者后继节点为取消等待状态，则从尾部向头部节点遍历，找到一个可以唤醒的合适节点
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);
}
// 设置节点为取消获取同步资源状态
/**
* Cancels an ongoing attempt to acquire.
*
* @param node the node
*/
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;
//如果当前节点是尾部节点，设置当前节点前置节点为尾部节点，并设置新的尾节点为Null
// 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;
// node不是尾部节点，也不是head节点后继节点时候，将node.next与node前置节点的next链接上
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 {
// node是头节点的后置节点的时候，将Node的后置节点唤醒，
unparkSuccessor(node);
}
// node的后继节点与队列断开，辅助系统gc
node.next = node; // help GC
}
}
//检测当前Node节点是否需要休眠
//休眠的条件：node的前置pre节点等待状态为signal：表明已经准备唤醒后置节点node
/**
* Checks and updates status for a node that failed to acquire.
* Returns true if thread should block. This is the main signal
* control in all acquire loops. Requires that pred == node.prev.
*
* @param pred node's predecessor holding status
* @param node the node
* @return {@code true} if thread should block
*/
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus;
if (ws == Node.SIGNAL) //前置节点已经设置成signal，准备唤醒后置节点，//后置节点Node可以休眠
/*
* 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.
*/
// 前置节点是取消等待状态，node节点不能进行休眠,将前置节点与Node解绑，并且不断找到非取消状态的前置节点，并与node重新绑定
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else {
/* //前置节点不是取消状态，signal状态，为0，或者propagete状态，设置前置节点为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;
}
---------------------------------------------------
-------独占模式获取同步资源方法----------------
---------------------------------------------------
/**
* Acquires in exclusive mode, ignoring interrupts. Implemented
* by invoking at least once {@link #tryAcquire},
* returning on success. Otherwise the thread is queued, possibly
* repeatedly blocking and unblocking, invoking {@link
* #tryAcquire} until success. This method can be used
* to implement method {@link Lock#lock}.
*
* @param arg the acquire argument. This value is conveyed to
* {@link #tryAcquire} but is otherwise uninterpreted and
* can represent anything you like.
*/
//请求同步资源，但是忽略等待请求资源过程中的中断（即不抛中断异常）
public final void acquire(int arg) {
if (!tryAcquire(arg) //尝试获取同步资源
&& acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) //尝试获取同步资源不成功则将当前线程入队列，返回是否入队列过程中线程发了中断
selfInterrupt(); 等待获取资源过程中线程发了中断，设置当前线程中断标志
}
//tryAcquire方法需要子类自行实现，如何获取锁，且获取是否成功
protected boolean tryAcquire(int arg) {
throw new UnsupportedOperationException();
}
// addWaiter方法单纯的将当前线程加入到等待队列中
//acquireQueued函数 为已经处于等待队列中线程获取同步资源，获取过程中不中断，只返回是否中断
final boolean acquireQueued(final Node node, int arg) {
boolean failed = true;
try {
boolean interrupted = false;
//利用死循环，不断为当前线程请求同步资源
for (;;) {
final Node p = node.predecessor(); //获取node的前置节点
//如果node节点的前置节点是头节点，则node节点属于头节点的后继节点，属于第一个有获取同步资源的资格的线程，立即尝试一次获取同步资源
if (p == head && tryAcquire(arg)) {
同步资源获取成功，node设置成头节点
setHead(node);
p.next = null; // help GC
failed = false;
return interrupted;
}
//获取失败的情况，检测下当前线程是否需要阻塞。需要阻塞就调用LockSupport.park(this); 将当前线程阻塞（等到调用unpark再继续执行获取资源代码），再判断是否有被中断过
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed) //如果在获取同步资源过程中，失败了，且还没获取到资源，就将当前线程从等待队列中移除，进行出队操作。
cancelAcquire(node);
}
}
//interrupt();方法将中断 线程的sleep/wait方法，不会中断正在执行的代码 获取同步资源，响应中断
public final void acquireInterruptibly(int arg)
throws InterruptedException {
if (Thread.interrupted()) // 如果当前线程中断标志为true,则抛出中断异常，和方法名相呼应
throw new InterruptedException();
if (!tryAcquire(arg)) // 尝试获取同步资源，如果没获取成功，则不断自旋获取
doAcquireInterruptibly(arg);
}
/tryAcquire(arg) // 尝试获取同步资源，子类实现，典型的摸板方法模式响应中断的进行获取同步资源
/**
* Acquires in exclusive interruptible mode.
* @param arg the acquire argument
*/
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())
//和忽略中断的请i求同步资源方法的唯一区别，就是等待资源过程中发生过中断，那么抛出中断异常
throw new InterruptedException();
}
} finally {
if (failed)
cancelAcquire(node);
}
}
// 在固定时间内获取同步资源，如果超时则获取失败
* Acquires in exclusive timed mode.
*
* @param arg the acquire argument
* @param nanosTimeout max wait time
* @return {@code true} if acquired
*/
private boolean doAcquireNanos(int arg, long nanosTimeout)
throws InterruptedException {
if (nanosTimeout <= 0L)
return false;
/ //获取资源最大时间
final long deadline = System.nanoTime() + nanosTimeout;
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;
}
nanosTimeout = deadline - System.nanoTime();
if (nanosTimeout <= 0L) //现在的时间已经最大的超时时间，返回获取失败
return false;
if (shouldParkAfterFailedAcquire(p, node) &&
nanosTimeout > spinForTimeoutThreshold)
// 如果需要休息了，则剩余等待时间（可休息的时间）是否大于配置值，大于则进行休息，小于就没必要休息了。
LockSupport.parkNanos(this, nanosTimeout);
if (Thread.interrupted())
throw new InterruptedException();
}
} finally {
if (failed)
cancelAcquire(node);
}
}
//释放在独占模式下获取到的同步资源
/**
* Releases in exclusive mode. Implemented by unblocking one or
* more threads if {@link #tryRelease} returns true.
* This method can be used to implement method {@link Lock#unlock}.
*
* @param arg the release argument. This value is conveyed to
* {@link #tryRelease} but is otherwise uninterpreted and
* can represent anything you like.
* @return the value returned from {@link #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;
}
----------------以上，就已经分析完了AQS给独占式同步资源操作的API方法
----------------下面开始，介绍AQS给共享式同步资源操作的API方法
//首先也是一个忽略中断的请求同步资源方法
/**
* Acquires in shared mode, ignoring interrupts. Implemented by
* first invoking at least once {@link #tryAcquireShared},
* returning on success. Otherwise the thread is queued, possibly
* repeatedly blocking and unblocking, invoking {@link
* #tryAcquireShared} until success.
*
* @param arg the acquire argument. This value is conveyed to
* {@link #tryAcquireShared} but is otherwise uninterpreted
* and can represent anything you like.
*/
public final void acquireShared(int arg) {
if (tryAcquireShared(arg) < 0) // 尝试获取共享资源，子类实现，判断是否成功与独占式不同，小于0代表获取失败，
doAcquireShared(arg); //获取失败，需要将当前线程加入等待队列
}
//非响应式中断，请求共享资源，和独占式获取逻辑类似
/**
* Acquires in shared uninterruptible mode.
* @param arg the acquire argument
*/
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);
}
}
//会对中断进行响应的，请求共享资源
public final void acquireSharedInterruptibly(int arg)
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
if (tryAcquireShared(arg) < 0)
doAcquireSharedInterruptibly(arg);
}
//响应中断的请求共享资源
/**
* Acquires in shared interruptible mode.
* @param arg the acquire argument
*/
private void doAcquireSharedInterruptibly(int arg)
throws InterruptedException {
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
for (;;) {
final Node p = node.predecessor();
if (p == head) {
int r = tryAcquireShared(arg);
if (r >= 0) {
setHeadAndPropagate(node, r);
p.next = null; // help GC
failed = false;
return;
}
}
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
// 当前线程在等待获取资源的过程中，中断过，则抛出中断异常来响应中断
throw new InterruptedException();
}
} finally {
if (failed)
cancelAcquire(node);
}
}
//在响应中断的基础上，加上等待的超时时间，获取同步资源
public final boolean tryAcquireSharedNanos(int arg, long nanosTimeout)
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
return tryAcquireShared(arg) >= 0 ||
//在允许的超时时间内，获取同步资源
doAcquireSharedNanos(arg, nanosTimeout);
}
//根据最大超时时间进行同步资源获取
/**
* Acquires in shared timed mode.
*
* @param arg the acquire argument
* @param nanosTimeout max wait time
* @return {@code true} if acquired
*/
private boolean doAcquireSharedNanos(int arg, long nanosTimeout)
throws InterruptedException {
if (nanosTimeout <= 0L)
return false;
final long deadline = System.nanoTime() + nanosTimeout;
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
for (;;) {
final Node p = node.predecessor();
if (p == head) {
int r = tryAcquireShared(arg);
if (r >= 0) {
setHeadAndPropagate(node, r);
p.next = null; // help GC
failed = false;
return true;
}
}
nanosTimeout = deadline - System.nanoTime();
if (nanosTimeout <= 0L)
return false;
if (shouldParkAfterFailedAcquire(p, node) &&
nanosTimeout > spinForTimeoutThreshold)
LockSupport.parkNanos(this, nanosTimeout);
if (Thread.interrupted())
throw new InterruptedException();
}
} finally {
if (failed)
cancelAcquire(node);
}
}、
//释放共享模式下的同步资源的
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) { //尝试释放同步资源，子类实现，模板模式体现
doReleaseShared(); // 执行资源释放操作
return true;
}
return false;
}
//执行共享模式下同步资源释放操作
/**
* Release action for shared mode -- signals successor and ensures
* propagation. (Note: For exclusive mode, release just amounts
* to calling unparkSuccessor of head if it needs signal.)
*/
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) {
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;
}
}
原文作者：服务端技术栈；如有侵权，请联系管理员删除。
