多线程并发编程7-AQS源码剖析
今天来说一说AQS，是AbstractQueuedSynchronizer抽象同步队列的简称。AbstractQueuedSynchronizer类是实现同步器的基础组件，并发包中的锁的底层使用的就是AQS。AbstractQueuedSynchronizer类中的唤醒和挂起是用的LockSupport类，并添加了一些别的参数，上层不同的锁机制以及同步机制就是对这些新增的参数进行不同的操作而形成的不同机制。下面就对AQS内部进行探秘。
AQS类中有如下几个重要的成员属性，就是对这些属性进行不同的操作组合形成了上层使用不同机制的锁以及同步机制。
FIFO双向的阻塞队列（关于什么是阻塞队列以及有什么作用后面会进行介绍）：
private transient volatile Node head：阻塞队列的头节点。
private transient volatile Node tail：阻塞队列的尾节点。
private volatile int state：同步状态，通过CAS算法进行设置。
class ConditionObject：内部类，可以直接访问AQS对象内部的变量，结合锁实现线程同步，每个ConditionObject对象对应一个单向链表队列，用来存放调用条件变量await方法后被阻塞的线程。这有个需要注意的，条件变量队列和上面说的阻塞队列是两个队列，需要区分开。
AQS类内部的函数可以分为两组，一组实现对独占方式进行获取和释放资源，一组则实现对共享方式进行获取和释放资源。那么什么是独占方式？什么是共享方式呢？
独占方式，就是说一个线程获取到了资源，就会标记该资源被当前线程获取，其他线程再尝试获取同个资源时就会发现该资源不是自己持有的，获取失败，则会被阻塞。例如JUC包下的独占锁ReentrantLock。
共享方式，共享方式的资源不和某个线程一一对应，当多个线程去请求获取资源时，一个线程获取到资源后，另一个线程再去获取同个资源，如果当前资源满足需要，则另一个线程也能获取到资源，相对于独占方式的资源和线程1对1的关系，共享方式的资源和线程是1对多的关系。例如JUC包下的Semaphore信号量。
下面对独占方式和共享方式获取和释放资源的源码进行讲解。
独占方式
获取资源
acquire(int arg)
获取独占资源。
public final void acquire(int arg) {
if (!tryAcquire(arg) && //(1)
acquireQueued( //(3)
addWaiter(Node.EXCLUSIVE), arg) //(2)
)
selfInterrupt();
}
（1）tryAcquire(arg)方法尝试获取资源，成功则直接返回。tryAcquire方法由上层的锁进行具体实现，不同机制的锁tryAcquire方法实现不同，例如JUC包下的ReentrantLock锁，分为公平锁和非公平锁，他们的tryAcquire方法实现就不同。
（2）tryAcquire(arg)方法尝试获取资源失败后，会将当前线程封装为类型为Node.EXCLUSIVE的node节点插入到AQS阻塞队列的尾部。
（3）阻塞队列中的线程获取资源，如果资源无法获取则会进行阻塞。
Node enq(final Node node)
插入node节点到AQS阻塞队列
private Node enq(final Node node) {
for (;;) {
//（1）
Node t =tail;
if (t ==null) {// Must initialize
if (compareAndSetHead(new Node()))
tail =head;
}else { //(2)
node.prev = t;
if (compareAndSetTail(t, node)) {
t.next = node;
return t;
}
}
}
}
（1）当尾节点为null，则创建一个哨兵节点，tail和head都被赋值为哨兵节点。
（2）当尾节点不为null, 则将当前node的pre节点设置为tail节点，使用 CAS算法把当前node设置为tail节点，并将旧的tail节点的next节点设置为当前node节点。
acquireQueued(final Node node, int arg)
阻塞队列中的线程获取资源，获取失败则进行阻塞。
final boolean acquireQueued(final Node node, int arg) {
boolean failed =true;
try {
boolean interrupted =false;
for (;;) {
final Node p = node.predecessor();
//(1)
if (p ==head && tryAcquire(arg)) {
setHead(node);
p.next =null; // help GC
failed =false;
return interrupted;
}
if (shouldParkAfterFailedAcquire(p, node) && //(2)
parkAndCheckInterrupt()) //(3)
interrupted =true;
}
}finally {
if (failed)
cancelAcquire(node); //(4)
}
}
（1）当一个node节点的pre节点为head节点，说明阻塞队列中当前节点为第一个节点，需要获取资源，则将该节点进行tryAcquire方法尝试获取资源，获取成功后设置head节点为node，并返回。
（2）检查和更新当前节点node的pre节点中线程的waitStatus状态。当线程需要被阻塞时返回true。
（3）执行LockSupport的Lock方法进行阻塞。
（4）取消正在进行的获取尝试。
shouldParkAfterFailedAcquire(Node pred, Node node)
检查和更新当前节点node的pre节点的waitStatus状态。当线程需要被阻塞时返回true
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus;
//(1)
if (ws == Node.SIGNAL)
return true;
//(2)
if (ws >0) {
do {
node.prev = pred = pred.prev;
}while (pred.waitStatus >0);
pred.next = node;
}else { //(3)
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}
waitStatus是记录当前线程等待状态，在解释源码之前先介绍一下waitStatus的几种状态：
1）CANCELLED（1）：线程被取消了。
2）SIGNAL（-1）：线程需要被唤醒。
3）CONDITION（-2）：线程在条件队列里面等待。
4）PROPAGATE（-3）：释放共享资源时需要通知其他节点。
（1）如果当前节点的pre节点的waitStatus是SIGNAL，则当前节点需要被阻塞，则返回true。
（2）如果当前节点的pre节点的waitStatus大于0，说明当前节点的pre节点被取消了，则进行循环找到前节点
的waitStatus小于等于0的，将前节点的waitStatus大于0的从阻塞队列中移除。
（3）如果当前节点的pre节点的waitStatus小于0，则使用CAS算法设置当前节点的pre节点的waitStatus为SIGNAL。下次再进入到shouldParkAfterFailedAcquire方法就会执行（1）的代码并返回true。
释放资源
release(int arg)
释放获取到的独占资源。
public final boolean release(int arg) {
if (tryRelease(arg)) { //(1)
Node h =head;
if (h !=null && h.waitStatus !=0)
unparkSuccessor(h); //（2）
return true;
}
return false;
}
（1）尝试释放资源，release方法由上层的锁进行具体实现，不同机制的锁release方法实现不同，例如JUC包下的ReentrantLock锁，分为公平锁和非公平锁，他们的release方法实现就不同。
（2）唤醒指定节点的后续节点。并不一定是唤醒当前节点的第一个后续节点，看下面的unparkSuccessor源码分析即可明白。
unparkSuccessor(Node node)
唤醒指定节点的后续节点。
private void unparkSuccessor(Node node) {
int ws = node.waitStatus;
//（1）
if (ws <0)
compareAndSetWaitStatus(node, ws, 0);
//（2）
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;
}
//（3）
if (s !=null)
LockSupport.unpark(s.thread);
}
（1）如果当前节点的waitStatus状态小于0，则使用CAS算法尝试设置当前节点waitStatus状态为0。
（2）如果当前节点的next节点为null，或者waitStatus状态>0(即被取消状态)，则从tail节点向head节点查找最后一个节点waitStatus状态小于0的节点。
（3）唤醒满足条件的节点中的线程。
共享方式
获得资源
acquireShared(int arg)
获取共享资源。
public final void acquireShared(int arg) {
//（1）
if (tryAcquireShared(arg) <0)
doAcquireShared(arg); //（2）
}
（1）尝试设置state值，设置成功则直接返回。tryAcquireShared方法由上层的锁进行具体实现，不同机制的锁tryAcquireShared方法实现不同，例如JUC包下的Semaphore信号量，分为公平和非公平方式，他们的tryAcquireShared方法实现就不同。
（2）调用tryAcquireShared方法设置state值失败后将当前线程封装为node实例插入阻塞队列，当满足阻塞条件的时候阻塞当前线程。
doAcquireShared(int arg)
共享方式获取资源
private void doAcquireShared(int arg) {
//（1）
final Node node = addWaiter(Node.SHARED);
boolean failed =true;
try {
boolean interrupted =false;
for (;;) {
//（1）
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;
}
}
//（2）
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted =true;
}
}finally {
//（3）
if (failed)
cancelAcquire(node);
}
}
（1）如果当前节点的pre节点为head节点，则调用tryAcquireShared方法尝试设置共享模式的state值，如果tryAcquireShared方法返回大于0，则将当前节点的后续节点中阻塞的节点唤醒。
（2）（3）和独占方式中的代码一样，具体查看上面介绍。
释放资源
releaseShared(int arg)
释放共享资源。
public final boolean releaseShared(int arg) {
//（1）
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}
（1）调用tryReleaseShared，尝试设置共享模式的state，具体的实现根据上层不同的功能会有不同的实现。tryReleaseShared返回大于0则调用doReleaseShared进行释放共享资源。
doReleaseShared()
释放共享资源，唤醒等待的线程。
private void doReleaseShared() {
//（1）
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;
}
}
（1）从head节点往tail节点进行遍历，唤醒所有阻塞的node。
不管是共享方式还是独占方式都存在一个带Interruptibly关键字的方法，这类方法表示当获取资源或获取资源失败被挂起时，其他线程中断了该线程，那个该线程会抛出InterruptedException异常而返回。
条件变量
在前面介绍了AQS类中有个ConditionObject内部类，用来存放调用条件变量await方法后被阻塞的线程，下面介绍ConditionObject类是如何支持作为条件变量使线程同步的。
下面先用一个例子来介绍什么是条件变量？如何使用？
ReentrantLock lock =new ReentrantLock(); //（1）
Condition condition = lock.newCondition(); //（2）
lock.lock(); //（3）
try {
condition.await(); //（4）
}catch (InterruptedException e) {
e.printStackTrace();
}finally {
lock.unlock(); //（5）
}
lock.lock(); //（6）
try {
condition.signal(); //（7）
}catch (Exception e) {
e.printStackTrace();
}finally {
lock.unlock(); //（8）
}
（1）创建锁对象。
（2）使用创建的锁对象创建条件变量，一个锁可以创建多个条件变量。
（3）获取锁，只有在获取锁的情况下才可以调用条件变量的await或signal方法，否则会抛出IllegalMonitorStateException异常。
（4）调用条件变量await方法，阻塞挂起当前线程。
（5）释放锁。
（6）（7）（8）唤醒一个调用条件变量的await方法而阻塞的线程，步骤和调用await一致。
条件变量是和锁进行关联，每个锁对象可以创建多个条件变量对象，每个条件变量对象内部都维护了一个条件队列，用来存放调用条件变量的await方法时被阻塞的线程，这个队列需要和AQS中的阻塞队列区分开。
当一个线程获取到锁后又调用了对应条件变量的await方法，则该线程会释放获取到的锁，并被转换为node节点插入到条件变量对应的条件队列里。
当另一个线程获取到锁后调用了对应条件变量的signal或signalAll方法，会把对应条件变量的条件队列里面的一个或全部node节点移动到AQS的阻塞队列里，等待获取锁。
下面看看具体源码的实现。
await()
public final void await()throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
Node node = addConditionWaiter(); //（1）
int savedState = fullyRelease(node); //（2）
int interruptMode =0;
//（3）
while (!isOnSyncQueue(node)) {
LockSupport.park(this);
if ((interruptMode = checkInterruptWhileWaiting(node)) !=0)
break;
}
//（4）
if (acquireQueued(node, savedState) && interruptMode !=THROW_IE)
interruptMode =REINTERRUPT;
//（5）
if (node.nextWaiter !=null)// clean up if cancelled
unlinkCancelledWaiters();
if (interruptMode !=0)
reportInterruptAfterWait(interruptMode);
}
（1）创建node对象，并插入到条件队列尾部。
（2）释放获取到的锁，如果当前线程没获取到锁资源，在释放锁的时候就会抛出IllegalMonitorStateException异常。
（3）当前node如果不在阻塞队列里则调用LockSupport.park阻塞当前线程，直到其他线程调用该条件变量的signal或signalAll方法，唤醒当前线程，则将当前node插入到阻塞队列中。
（4）当前node已经插入阻塞队列，需要等待获取到锁而被阻塞。
（5）将当前node从条件队列中移除。
signal()
唤醒一个调用相同条件变量await方法而被阻塞的线程。
public final void signal() {
//（1）
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
Node first =firstWaiter;
if (first !=null)
//（2）
doSignal(first);
}
（1）判断当前线程是否回去到锁资源，如果没有获取到锁资源则抛出IllegalMonitorStateException异常。
（2）从条件队列中从头到尾查找第一个waitStatus的状态为Node.CONDITION的节点，将该节点waitStatus的状态修改为0，并插入到阻塞队列，之后唤醒该node对应的线程。
今天的分享就到这，有看不明白的地方一定是我写的不够清楚，所有欢迎提任何问题以及改善方法。
