深入理解Condition
我们知道在线程的同步时可以使一个线程阻塞而等待一个信号，同时放弃锁使其他线程可以能竞争到锁。在synchronized中我们可以使用Object的wait()和notify方法实现这种等待和唤醒，但是在Lock中怎么实现这种wait和notify呢？答案是Condition，学习Condition主要是为了方便以后学习blockqueue和concurrenthashmap的源码，同时也进一步理解ReentrantLock。
首先，Condition是一个接口，主要有以下的方法：
Paste_Image.png
截取一段Condition的注释，也是它的用法：
*class BoundedBuffer {
* <b>final Lock lock = new ReentrantLock();</b>
* final Condition notFull = <b>lock.newCondition(); </b>
* final Condition notEmpty = <b>lock.newCondition(); </b>
*
* final Object[] items = new Object[100];
* int putptr, takeptr, count;
*
* public void put(Object x) throws InterruptedException {
* <b>lock.lock();
* try {</b>
* while (count == items.length)
* <b>notFull.await();</b>
* items[putptr] = x;
* if (++putptr == items.length) putptr = 0;
* ++count;
* <b>notEmpty.signal();</b>
* <b>} finally {
* lock.unlock();
* }</b>
* }
*
* public Object take() throws InterruptedException {
* <b>lock.lock();
* try {</b>
* while (count == 0)
* <b>notEmpty.await();</b>
* Object x = items[takeptr];
* if (++takeptr == items.length) takeptr = 0;
* --count;
* <b>notFull.signal();</b>
* return x;
* <b>} finally {
* lock.unlock();
* }</b>
* }
* }
主要是ReentrantLock的实例lock调用的newCondition()方法:
public Condition newCondition() {
return sync.newCondition();
}
此方法生成了一个新的Condition实例，这也就是说在多线程共享一个Lock的时候，只能调用一次newCondition()使所有的线程共享此Condition。
调用Sync的newCondition()。Sync是AbstractQueuedSynchronizer的子类：
final ConditionObject newCondition() {
return new ConditionObject();
}
ConditionObject是AbstractQueuedSynchronizer的内部类,ConditionObject的实现代码很长，就不贴出来了。
在获取到了Condition实例之后，在获取了lock之后，调用Condition的await()方法，可以实现当前线程阻塞等待，同时放弃锁的效果。await()方法的实现如下：
/**
* Implements interruptible condition wait.
* <ol>
* <li> If current thread is interrupted, throw InterruptedException.
* <li> Save lock state returned by {@link #getState}.
* <li> Invoke {@link #release} with
* saved state as argument, throwing
* IllegalMonitorStateException if it fails.
* <li> Block until signalled or interrupted.
* <li> Reacquire by invoking specialized version of
* {@link #acquire} with saved state as argument.
* <li> If interrupted while blocked in step 4, throw InterruptedException.
* </ol>
* tips：调用await，当前线程在此Condition上等待，同时将此线程实例化成一个Node，加入等待队列中
*
*
*/
public final void await() throws InterruptedException {
//判断是否中断
if (Thread.interrupted())
throw new InterruptedException();
//在此Condition维护的等待队列中增加节点
Node node = addConditionWaiter();
//此处做锁的释放，将当前的线程占用的Lock释放，通过修改Lock的State实现
int savedState = fullyRelease(node);
int interruptMode = 0;
//判断当前的线程是否在Syn队列中，此处isOnSyncQueue方法实现的功能比较多，而且不是Condition的方法，怀疑是一个方法做了多个事情。
while (!isOnSyncQueue(node)) {
//如果当前的node的WaitStatus是Condition，那么park此线程
LockSupport.park(this);
if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
break;
}
//线程唤醒之后继续争抢Lock
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
interruptMode = REINTERRUPT;
if (node.nextWaiter != null) // clean up if cancelled
unlinkCancelledWaiters();
if (interruptMode != 0)
reportInterruptAfterWait(interruptMode);
}
首先是检查中断。如果中断，那么抛出中断异常。
如果没有中断，调用addConditionWaiter（）:
/**
* Adds a new waiter to wait queue.
* @return its new wait node
*
* tips： 在Condition的队列中增加一个Node节点，代表当前Wait的线程，等待Signal信号唤醒。
* 此处Node的首尾节点都没有用Volatile修饰，是因为到await方法的线程都是获取锁的，利用锁已经能保证了
* Node的可见性了。也就是说，此方法本身就是线程安全的，这里就没必要利用Volatile保证可见性。
*/
private Node addConditionWaiter() {
Node t = lastWaiter;
// If lastWaiter is cancelled, clean out.
//遍历整个双向链表，清除已经取消的节点
if (t != null && t.waitStatus != Node.CONDITION) {
unlinkCancelledWaiters();
t = lastWaiter;
}
Node node = new Node(Thread.currentThread(), Node.CONDITION);
if (t == null)
firstWaiter = node;
else
t.nextWaiter = node;
lastWaiter = node;
return node;
}
这里主要是构建了一个等待队列的双向链表，我们叫它等待队列，新建一个waitStatus为CONDITION的Node,并加到队尾。这里没有用volatile遍历保证Node的可见性，但是因为默认是已经获取锁，所以方法默认是线程安全的。
然后我们需要释放当前的锁，如果当前锁不释放，那就死锁了，so调用fullyRelease（）：
/**
* Invokes release with current state value; returns saved state.
* Cancels node and throws exception on failure.
* @param node the condition node for this wait
* @return previous sync state
*/
final int fullyRelease(Node node) {
boolean failed = true;
try {
int savedState = getState();
if (release(savedState)) {
failed = false;
return savedState;
} else {
throw new IllegalMonitorStateException();
}
} finally {
if (failed)
node.waitStatus = Node.CANCELLED;
}
}
这里主要是释放锁的逻辑。实现主要是在Release方法中，Release方法在unLock的时候也会调用到，主要是释放线程占用的锁和唤醒同步队列中阻塞的线程：
/**
* 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;
}
这段代码在之前的《深入理解ReentrantLock》中写过了，就不分析了。
接下来是判断当前的节点是不是在同步队列中，这里可能会有一点疑问，就是为什么是判断同步队列而不是等待队列呢？
其实在后面signal的时候会有一个将node转入到同步队列的过程，从而使线程重新加入到锁竞争中。这里判断是否在同步队列中，其实就是判断是不是已经signal：
/**
* Returns true if a node, always one that was initially placed on
* a condition queue, is now waiting to reacquire on sync queue.
* @param node the node
* @return true if is reacquiring
* tips：此处第一个判断当前Node的状态，如果是CONDITION，表明是正常的没有被唤醒的节点，返回false 表名当前的节点还没进入到Syn队列
* 第二个判断做了个取巧，从而避免遍历这个歌Syn队列，直接判断Node的下一个节点是否为null，不为null表明在Syn队列中
* 第三个判断遍历整个Node的Syn队列，如果存在返回true
* 这里有一个Wait队列到Syn队列的转化过程
*/
final boolean isOnSyncQueue(Node node) {
if (node.waitStatus == Node.CONDITION || node.prev == null)
return false;
if (node.next != null) // If has successor, it must be on queue
return true;
/*
* node.prev can be non-null, but not yet on queue because
* the CAS to place it on queue can fail. So we have to
* traverse from tail to make sure it actually made it. It
* will always be near the tail in calls to this method, and
* unless the CAS failed (which is unlikely), it will be
* there, so we hardly ever traverse much.
*/
return findNodeFromTail(node);
}
//判断当前的线程是否在Syn队列中,判断节点是不是在同步队列中，因为上一步已经释放了锁，也就是说此时可能有线程已经获取锁同时可能已经调用了singal方法，如果已经唤醒，那么就不应该park了，而是退出while方法，从而继续争抢锁
while (!isOnSyncQueue(node)) {
LockSupport.park(this);
if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
break;
}
如果之前返回了true，那么表示已经在队列中，那么继续争抢锁，这种主要场景是已经释放锁，但是还没有park线程的时候，线程被singal。如果返回false，证明没有singal，所以要park当前的线程。
这样await的逻辑就清晰了，接下来是一个singal的过程，调用singal的时候：
/**
* Moves the longest-waiting thread, if one exists, from the
* wait queue for this condition to the wait queue for the
* owning lock.
*
* @throws IllegalMonitorStateException if {@link #isHeldExclusively}
* returns {@code false}
*/
public final void signal() {
//检查当前线程是不是，Lock占用的线程
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
//取出Condition队列中第一个Node
Node first = firstWaiter;
if (first != null)
//唤醒第一个Node
doSignal(first);
}
首先判断当前线程是不是持有锁的先，如果不是，那么锁没意义，抛出异常。如果是，那么去除等待队列中首节点，调用doSignal唤醒：
/**
* Removes and transfers nodes until hit non-cancelled one or
* null. Split out from signal in part to encourage compilers
* to inline the case of no waiters.
* @param first (non-null) the first node on condition queue
*/
private void doSignal(Node first) {
do {
if ( (firstWaiter = first.nextWaiter) == null)
lastWaiter = null;
first.nextWaiter = null;
} while (!transferForSignal(first) &&
(first = firstWaiter) != null);
}
这里主要是遍历整个等待队列，知道找到一个可唤醒的节点，所以主要关心唤醒逻辑，即transferForSignal：
/**
* Transfers a node from a condition queue onto sync queue.
* Returns true if successful.
* @param node the node
* @return true if successfully transferred (else the node was
* cancelled before signal).
*/
final boolean transferForSignal(Node node) {
/*
* If cannot change waitStatus, the node has been cancelled.
* 如果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;
}
这里其实是有疑惑的，我开始也是没看明白，后来想了想才明白。这里主要的疑惑是!compareAndSetWaitStatus(p, ws, Node.SIGNAL)一致false的，所以一直不会执行LockSupport.unpark(node.thread)。那也就是说无法唤醒await的线程？其实不是的，因为唤醒操作时在unlock或者是await时进行的。因为在release的方法中会做同步队列中节点的唤醒，而signal之后的线程是加入到了同步队列中的，也就是说signal之后的线程还是需要重新竞争锁的，而不是直接获取锁。
这一点和synchronized的设计不谋而合，因为在synchronized中调用notify也不是及时的获取锁的，而是要等待synchronized块结束之后才竞争。
ReentrantLock还是有很多的优秀实现思想，比如CAS，转换严格的双向链表队列等。后面我还会继续学习总结，不断进步。
