JUC包之AbstractQueuedSynchronizer源码学习

AQS是JUC包的基础，几乎所有的JUC工具类，都是用了AQS进行实现，AQS使用了CLH lock的队列作为基础，关于CLH，前文有一篇转载的又说到什么事CLH lock，AQS中的CLH,其实就是一个FIFO的队列，队列中的每个结点（线程）只要等待其前继释放锁就可以了。这是JDK内画的 一张图：

* <pre>
*      +------+  prev +-----+       +-----+
* head |      | <---- |     | <---- |     |  tail
*      +------+       +-----+       +-----+
* </pre>

这张图简单的说明了AQS内部阻塞队列的实现，这里省略了next指向下一节点的指针，只是标识出了pre指向前继的指针。

要看AQS，还要先从内部Node节点类说起。

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;

/** waitStatus value to indicate thread has cancelled */
static final int CANCELLED = 1;
/** waitStatus value to indicate successor's thread needs unparking */
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;

volatile int waitStatus;

volatile Node prev;

volatile Node next;

volatile Thread thread;

Node nextWaiter;

final boolean isShared() {
return nextWaiter == SHARED;
}

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

Node保留了前继prev和后继next指针，以及对象Thread，同时有两个Node类型的标记值SHARED和EXCLUSIVE标记模式是互斥的还是共享的。我们还可以看到一个waitStatus字段，默认有四种值：

CANCELLED = 1 说明此节点对应的线程已经取消

SIGNAL = -1 说明后继节点的线程（successor's thread）需要unparking

CONDITION = -2，说明此线程正在某个条件上阻塞

PROPAGATE = -3，下个共享获取请求必须无条件通过

Node用来作为queue队列节点，同时存储了阻塞的线程，并且用不同的状态和字段标识更多的信息。AQS队列是一个FIFO先进先出的队列。

我们再来看下AQS类自身，

<div>  private transient volatile Node head;</div><div>
</div><div>    /**
     * Tail of the wait queue, lazily initialized.  Modified only via
     * method enq to add new wait node.
     */
    private transient volatile Node tail;</div><div>
</div><div>    /**
     * The synchronization state.
     */
    private volatile int state;</div>

这是AQS自身的三个属性，其中state尤为重要，很多工具类都是对state的值进行不同定义而实现不同的功能的。比如信号量和可重入互斥锁等。

这是节点入队方法，不用细说：

/**
* 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);
// 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;
}

AQS还有一个很重要的工具类，LockSupport，后面我们会用到，关于此类，可以参考博客http://www.cnblogs.com/zhanjindong/p/java-concurrent-package-aqs-locksupport-and-thread-interrupt.html。

我们看下AQS下的unparkSuccessor(Node node)方法：

//Wakes up node's successor, if one exists. /*当prev节点为signal时，说明下一个节点需要被unparking，这此方法并没有限制为signal，直接unparking*/

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)//若状态值为负数，则置为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) {//下个节点为null或状态值>0，则从tail节点开始向前遍历，找到第一个状态值<=0的，并unpark此节点的线程
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);
}

看下互斥下阻塞获取方法：

public final void acquire(int arg) {
if (!tryAcquire(arg) &&   //addWaiter将创建新节点，添加到阻塞队列
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();//调用线程的中断方法
}

其中tryAcquire方法只是抛出了异常，强制子类去实现此方法，AQS保留了

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

等方法去让子类去实现。

tryAcquire(arg)获取失败后， 将线程添加到阻塞队列。

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

在调用addWaiter将线程添加到阻塞队列之后，进行recheck，获取node的前继，若是head节点且tryAcquire获取资源成功，则将head出队列，并将此节点设置为head节点。

/**
* 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，说明后继可以等待unparking，将后继park
/*
* This node has already set status asking a release
* to signal it, so it can safely park.
*/
return true;
if (ws > 0) {//前继已经取消，往前遍历，知道状态！>0,将遍历后的节点next指向此节点，去掉刚才的已取消节点
/*
* 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，重置为signal，让后继通过前继的signal，表明自己等待unparking
/*
* 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;
}

private final boolean parkAndCheckInterrupt() {
LockSupport.park(this);
return Thread.interrupted();//将线程的中断状态返回并且清零
}

继续看下释放操作：

/**
* 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)) {//如果tryRelease释放成功，则将head出队
Node h = head;
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);//找到满足条件的最靠近head的后继节点并unpark
return true;
}
return false;
}