Java并发之AQS原理
本文是Java并发专题（欢迎大家关注）的一篇。
以下是完整的目录：
Java并发之基础知识
Java并发之volatile关键字
Java并发之synchronized关键字
Java并发之原子类
Java并发之线程池
Java并发之并发工具类
Java并发之AQS原理
一.总体框架
AQS是指AbstractQueuedSynchronizer。它是一个抽象类，java并发包里的ReentrantLock、CountDownLatch和Semaphroe等重要的工具类都是基于AQS来实现的。
总体来说，AQS维护了一个volatile的state变量代表共享资源，还有一个FIFO的等待队列，在多线程争夺资源被阻塞时会进入此退了。等待队列是个双向链表记录则没有获取的执行许可的线程。等待队列中的结点元素是AQS自定义的static的内部类Node。AQS支持共享和独占两种模式。ReentrantLock就是独占型的，只有一个线程可以获得到锁并执行。CountDownLatch和Semaphore就是共享型，允许多个线程同时执行。
AQS是一个抽象类，并不能被直接实例化使用。它的作用是提供等待队列的管理，包括如何入队何时唤醒等。而具体的资源如何获取和释放等由具体的自定义同步器来实现。也就是说ReentrantLock等类自定义了资源(state)的获取和释放，而使用AQS的来管理阻塞队列。不同的自定义资源获取方式实现了CountDownLatch和Semaphore等类。
自定义同步方法需要实现的方法有：
- isHeldExclusively() //返回该线程是否正在独占资源，只有用的condition才需要去实现它
- tryAcquire(int); //独占方式，尝试获取资源，成功返回true，失败返回false
- tryRelease(int); //独占方式，尝试释放资源，成功返回true，失败返回false
- tryAcquireShared(int); //共享方法，尝试获取资源。返回负数表示失败，0表示成功，但没有可用资源了，正数表示成功且有剩余资源
- tryReleaseShared(int);//共享方式。尝试释放资源。如果释放后运行唤醒后续结点返回true，否则返回false
这其中tryAcquire和tryRelease是一组，用于实现独占资源的情况，如ReentrantLock；tryAcquireShared和tryReleaseShared是一组用于实现共享资源的情况，如CountDownLatch。
二.源码分析
2.1 acquire方法源码详解
在AQS中一个重要的方法是acquire(int)，这个方法实现请求资源和阻塞线程的功能。下面先贴一下它的源码：
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
这个方法里只有一个if语句。首先执行tryAcquire(int)方法，前面说了这个方法需要子类来自定义，这里先看一下它的代码：
protected boolean tryAcquire(int arg) {
throw new UnsupportedOperationException();
}
可以看到在AQS中tryAcquire方法直接抛出了异常。因为具体的获取资源细节需要子类根据自己要实现的功能来写，AQS只负责阻塞队列的管理等工作。同时注意到这个方法并不是一个抽象的方法。其实前面说的需要子类实现的5个方法都不是抽象的，因为子类并不一定需要实现所有这些方法，这提供了一定的灵活性。
2.1.1 addWaiter方法详解
先忙接着看acquire方法。在if语句里，如果tryAcquire返回true，那么acquire就返回了，说明成功获取到了资源。如果tryAcquire返回false，if语句的前半句判断就成立了，需要继续执行&&右边的acquireQueued方法，执行它之前先执行了addWaiter。先看一下addWaiter它的代码：
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;
}
可以看到这个方法有一个Node参数。Node类便是aqs维护的FIFO队列中的元素的类型。回顾一下acquire()方法的代码，是将Node.EXCLUSIVE作为参数传入了addWaiter。查看Node源码发现有这么一句：
/** Marker to indicate a node is waiting in exclusive mode */
static final Node EXCLUSIVE = null;
原来这是一个null值，用来表示独占性线程。不管如何，先继续看addWaiter的源码吧。
第一句代码：Node node = new Node(Thread.currentThread,mode);新建了一个表示当前线程的结点。刚才传入的null作为模式传给构造方法。进入对象构造方法查看：
Node(Thread thread, Node mode) { // Used by addWaiter
this.nextWaiter = mode;
this.thread = thread;
}
继续看addWaiter的后续代码，发现是获取了当前队列的尾节点，并将新建结点的prev指针执行尾节点，再使用cas尝试替换尾节点，如果成功，那么当前结点就成为新的尾节点，返回。
如果cas失败或者当前tail为null，调用eng方法处理。下面看一下eng的代码：
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;
}
}
}
}
熟悉AtomicInteger的朋友看到这段代码一定会感动非常熟悉。这里就是使用了循环尝试的方式来进行cas操作，指导成功为止。另外当tail==null时，先新建head结点再进行操作，当前这里给head变量反之也是使用了cas操作。
综上，addWaiter()进行的操作就是安全地更新队列的tail指针。
2.1.2 acquireQueue方法详解
下面继续看acquire()方法。再把代码贴一次。
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
给acqureQueued方法传入的第一个参数是addWaiter方法的返回值，回想一下刚才的addWaiter方法，发现它的返回值是新创建的表示当前线程的Node结点。acquireQueued方法的另一个参数是acquire的形参arg，这个一般是获取资源的个数，像ReentrantLock的lock方法就是调用了acquire(1)。下面看一下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;
return interrupted;
}
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}
这个方法的主体是一个死循环，不断测试两件事：1.是否是头结点的下一个节点，说明该轮到自己获取资源了。2：是否可以休息了。判断1成功后就用tryAcquire获取资源，成功后设置当前结点为头结点，返回。如果1判断不成功则执行shouldParkAfterFailedAcquire方法，先贴一下代码：
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.
*/
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}
这个方法的工作是找到当前结点之前的一个未取消的结点，将其waitStatue改为SIGNAL（-1）。这样在该结点释放资源时就会唤醒当前结点。
当shouldParkAfterFailedAcquire返回true之后，当前线程就可以去休息了——调用parkAndCheckInterrupt方法：
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this);
return Thread.interrupted();
}
这个方法使用了LockSupport的park方法，使线程进入waiting状态。当其它线程调用unPark方法，或此线程被中断后才会返回。
2.1.3小结
下面来总结一下acquire方法。
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
先尝试获取资源，获取到的情况直接返回。获取不到将线程加入队列：首先将tail指向表示当前线程的结点，使用CAS操作更新tail。之后执行acquireQueued方法，如果是当前队列的第二个则再次尝试获取tryAcquire，成功后将自己设置为head（head表示已经获取到的资源的结点）。不能获取资源时判断是否可以park()，判断依据是其prev的结点的waitState是否是signal，即是否会在释放资源时通知它。之后当前线程调用park进入waiting状态。waitting结束时返回是否中断标志，并重置标志。回到acquire，如果waitting期间中断过，则调用selfInterrupt响应中断。
2.2 release(int)
此方法是独占模式下释放资源的顶层方法。
public final boolean release(int arg) {
if (tryRelease(arg)) {
Node h = head;
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}
这里可以看出释放资源成功时，获取到head结点（因为head结点表示的线程就是当前获取到资源的线程），执行unparkSuccessor()操作。这里便和shouldParkAfterFailedAcquire中‘休息’的代码相呼应。如果那里设置了waitStatus为signal就会使用LockSupport.unpark方法来唤醒等待的线程。
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或等待状态>0（当前结点被取消），则从后往前找到正在应该被唤醒的结点
*/
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);
}
对了，tryRelease方法也是具体的同步器来实现的。
2.3 其它方法
acquireShared(int)和releaseShared()方法是共享模式下获取资源和释放资源的方法。这里不再详细展开了，请看参考资料里的文章。
参考资料：1.Java并发之AQS详解
