AQS之Condition
vOoT 人气:2一、引言
一般我们在使用锁的Condition时，我们一般都是这么使用，以ReentrantLock为例，
ReentrantLock lock = new ReentrantLock();
Condition condition = lock.newCondition();
lock.lock();
try{
condition.await();
}finally{
lock.unlock();
}
lock.lock();
try{
condition.signal();
}finally{
lock.unlock();
}
从上面可以知道，我们调用Condition的await和signal方法必须是在获取得到锁的情况下，首先我们以这个为基础，先不管是如何获取得到锁的，那么上面的程序在condition.await()时阻塞当前调用的线程，而调用 condition.signal()方法的时候可能唤起一个正在await阻塞的线程，我这里说的是可能不是一定。为什么这么说，我们来看下await()方法主要做了什么事情。
二、分析
下面是await 方法的在jdk8的源码，
下面是await 方法的在jdk8的源码，
public final void await() throws InterruptedException {
if (Thread.interrupted()) // ①
throw new InterruptedException();
Node node = addConditionWaiter(); //②
int savedState = fullyRelease(node); //③
int interruptMode = 0;
while (!isOnSyncQueue(node)) { //④
LockSupport.park(this);
if ((interruptMode = checkInterruptWhileWaiting(node)) != 0) // ⑤
break;
}
if (acquireQueued(node, savedState) && interruptMode != THROW_IE) //⑥
interruptMode = REINTERRUPT;
if (node.nextWaiter != null) // clean up if cancelled ⑦
unlinkCancelledWaiters();
if (interruptMode != 0) // ⑧
reportInterruptAfterWait(interruptMode);
}
执行过程如下，
① ，判断当前显示是否被interrupt? 是的话，会抛出中断异常InterruptedException,
② ，调用addConditionWaiter方法，主要是new 一个以当前线程为数据的节点Node,然后添加到condition条件队列里。
③ ，调用fullyRelease方法，该方法内部调用release方法，而release方法主要是在AQS队列里面唤起第一个节点（即head的next节点）的线程（如果head后继节点存在的话）。这时，在ASQ同步器内至少有2个活动的线程（一个是当前线程(可能是头节点)，另一个是唤起的线程（如果存在））。如果唤起失败，会抛异常IllegalMonitorStateException。注：当存在唤起的线程的时候，这个线程就可以去争取获取锁。
④ ，通过isOnSyncQueue方法判断该节点是否在AQS队列中，当调用await时候，肯定不在AQS上，因为addConditionWaiter方法是new 一个新的Node.接着会进入while循环里面。调用 LockSupport.park(this);阻塞当前线程，相当于释放锁。
⑤ ，当当前线程被唤起的时候（可能是 另一个await线程唤起的第一个节点可能是这个线程，或者在signal下，前驱节点已经cancel时，第一个firstWaiter节点是该当前节点），需要判断是否被中断，存储在interruptMode，如果被中断则break,否则checkInterruptWhileWaiting返回0，那么会接着判断node节点是否在AQS中，如果还是不在的话，park当前线程，否则跳出while循环。那么node节点是什么时候被加入到AQS上的，答案是在signal方法上。
⑥ ，当node节点在AQS队列时，我们需要获取锁，只有当前线程的节点Node在AQS队列上，才能去争取锁。争取锁就是通过调用acquireQueued方法。等下来分析下acquireQueued方法。
⑦ ， 如果当前节点的node.nextWaiter不为空，说明还有其他线程在该condition上，并且当前的线程已经获取锁，接着清除条件队列上的cancel类型节点
⑧ ， 如果interruptMode 是InterruptedException类型或者REINTERRUPT类型。则进行相应的抛中断异常或者线程自我中断标志位设置。
接着，来分析下signal方法
public final void signal() {
if (!isHeldExclusively()) // ①
throw new IllegalMonitorStateException();
Node first = firstWaiter;
if (first != null)
doSignal(first); // ②
}
执行过程如下，
① ，如果当前线程不是持有该condition的锁，那么执行抛IllegalMonitorStateException异常。
② ，调用doSignal方法，并且条件队列的首节点传入。
private void doSignal(Node first) {
do {
if ( (firstWaiter = first.nextWaiter) == null) // ①
lastWaiter = null;
first.nextWaiter = null;
} while (!transferForSignal(first) && (first = firstWaiter) != null); // ②
}
① ，这种条件队列firstWaiter指针为next节点，因为当前的节点需要被移除条件队列，并且next节点为空，那么lastWaiter置为null,说明是空条件队列，接着把first.nextWaiter=null,说明移除了条件队列
② ，在这里有2步操作，一是transferForSignal,二是 first = firstWaiter，如果我们当前first节点入AQS队列成功，那么transferForSignal返回true,则doSignal的while循环结束，
如果当前的first节点入AQS返回失败，则需要next的节点重新signal，保证有一个成功的firstWaiter节点入AQS队列。接着来看下transferForSignal 方法主要做了什么事情。
final boolean transferForSignal(Node node) {
/*
* If cannot change waitStatus, the node has been cancelled.
*/
if (!compareAndSetWaitStatus(node, Node.CONDITION, 0)) // ①
return false;
/*
* Splice onto queue and try to set waitStatus of predecessor to
* indicate that thread is (probably) waiting. If cancelled or
* attempt to set waitStatus fails, wake up to resync (in which
* case the waitStatus can be transiently and harmlessly wrong).
*/
Node p = enq(node); // ②
int ws = p.waitStatus;
if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL)) // ③
LockSupport.unpark(node.thread);
return true;
}
① ，首先传入的节点node是条件队列的第一个节点（在外部已经移除），改变其状态CONDITION，为初始状态0,如果改变失败，说明该节点已经不是条件节点，直接返回false,doSignal方法重新调用新firstWaiter节点入AQS队列，
② ，把首节点入AQS节点,enq（）方法返回的是入节点的前驱节点。从这里核心方法可以知道，signal() 方法的作用其实只是把等待队列中第一个非取消节点转移到AQS的同步队列尾部。转移后的节点很可能正在在同步队列阻塞着，什么时候唤醒，取决于它的前驱节点是否是头节点。
③ ，如果当前前驱节点的waitStatus>0(说明是CANCELLED状态)，前驱节点已经Canncel（说明前驱节点已经中断等情况），则可以调用LockSupport.unpark(node.thread)唤起线程，则await方法的park返回可以立即返回，预先将AQS同步队列中取消的节点移除掉，而不用等到获取同步状态失败的时候再去判断了，起到一定的优化作用。
最后来分析下获取锁方法 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);
}
}
① ，传入node 为需要获取锁的节点，arg为之前state状态.
② ，如果传入的节点的前驱节点是head节点，说明当前节点是AQS队列的首节点，可以尝试去获取锁，即我们需要是要实现的同步语义方法tryAcquire，如果同步语义获取锁成功，则设置当前头节点为头节点。
这里注意返回的值得语义是是否发生中断，而不是获取锁是否成功。
③ ，调用 shouldParkAfterFailedAcquire方法，该方法用来判断获取锁失败后是否需要park当前线程，如果需要park线程，则接着判断该线程是否有中断标志。
接着我们来看下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;
}
① 前驱节点pred的waitStatus为SIGNAL，说明当前节点有效，返回true,代表需要park当前线程，
② 前驱节点已经取消，则删除去取消掉的前驱节点，返回false，外面继续for循环获取锁
③ 处在该条件语句的前驱节点的waitStatus必定是 0,或者是传播PROPAGATE，则设置传播节点为SIGNAL，然后返回false,则接着去for循环获取锁，并且失败的时候，调用shouldParkAfterFailedAcquire时知道前驱为SIGNAL（之前由③设置），则需要park线程。
从这里可以知道transferForSignal方法中，!compareAndSetWaitStatus(p, ws, Node.SIGNAL)语句，如果对其前驱节点设置Node.SIGNAL失败，则不需要等到acquireQueued去判断是否需要park线程，直接unpark线程即可
加载全部内容