Condition定义了等待/通知两种类型的方法，当前线程调用这些方法时，需要提前获取到Condition关联的锁。
Condition对象由Lock对象(调用Lock的newCondition()方法)创建出来的，换句话说，Condition是依赖Lock对象的。
首先跟一下Condition的await方法:
await
需要注意的是：调用await方法的线程肯定是持有锁的线程，即同步队列的头节点。
public final void await() throws InterruptedException {
// 如果当前线程被中断，抛出中断异常
if (Thread.interrupted())
throw new InterruptedException();
// 将当前线程添加到Condition条件队列
Node node = addConditionWaiter();
// 当前线程释放掉持有的锁，并将锁之前持有的状态值保存下来
int savedState = fullyRelease(node);
int interruptMode = 0;
while (!isOnSyncQueue(node)) {
// 若当前线程不在同步队列中，则挂起该线程
// 被唤醒的时机有2个：
// 1. 当前线程被中断；
// 2. 当前线程被signal
LockSupport.park(this);
// 当前线程被唤醒，检查该线程是否在挂起期间被中断过，若被中断过，跳出while循环
if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
break;
}
// 关于中断模式：
// 1. 当在被通知前被中断则将中断模式设置为THROW_IE;
// 2. 当在被通知后则将中断模式设置为REINTERRUPT(因为acquireQueued不会响应中断)
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
interruptMode = REINTERRUPT;
if (node.nextWaiter != null) // clean up if cancelled
unlinkCancelledWaiters();
if (interruptMode != 0)
reportInterruptAfterWait(interruptMode);
}
接着跟一下各个子方法：
addConditionWaiter
addConditionWaiter用于将当前线程添加进Condition条件队列:
private Node addConditionWaiter() {
// 条件队列尾节点
Node t = lastWaiter;
// 如果尾节点线程已取消，则执行unlinkCancelledWaiters进行取消节点的清除
if (t != null && t.waitStatus != Node.CONDITION) {
unlinkCancelledWaiters();
t = lastWaiter;
}
// 将当前线程封装为Node节点，状态为Node.CONDITION
Node node = new Node(Thread.currentThread(), Node.CONDITION);
// 如果尾节点为null，证明条件队列为空，则将头节点指针指向当前节点
if (t == null)
firstWaiter = node;
// 当前节点作为尾节点的后继节点
else
t.nextWaiter = node;
// 将尾节点指针指向当前节点
lastWaiter = node;
return node;
}
fullyRelease
fullyRelease用于释放当前线程持有的锁。
final int fullyRelease(Node node) {
boolean failed = true;
try {
// 获取当前线程持有锁的状态值
int savedState = getState();
// 释放锁，并返回savedState
if (release(savedState)) {
failed = false;
return savedState;
} else {
throw new IllegalMonitorStateException();
}
} finally {
// 若最终失败，将当前节点的状态修改为Node.CANCELLED
if (failed)
node.waitStatus = Node.CANCELLED;
}
}
checkInterruptWhileWaiting
checkInterruptWhileWaiting用于检查线程挂起期间的中断状态。
// 检查线程挂起期间的中断状态：
// 1. 若在被signal之前中断，则返回THROW_IE
// 2. 若在被signal之后中断，则返回REINTERRUPT
private int checkInterruptWhileWaiting(Node node) {
return Thread.interrupted() ?
(transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) :
0;
}
// 如果线程在被signal之前cancell，则返回true
final boolean transferAfterCancelledWait(Node node) {
if (compareAndSetWaitStatus(node, Node.CONDITION, 0)) {
enq(node);
return true;
}
while (!isOnSyncQueue(node))
Thread.yield();
return false;
}
signal
signal方法将会唤醒在条件等待队列中等待时间最长的节点(首节点)，在唤醒节点之前，会将节点移动到同步队列中。
public final void signal() {
// 如果当前线程不是持有该condition的锁，抛IllegalMonitorStateException异常
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
Node first = firstWaiter;
// 如果条件队列不为空，执行doSignal
if (first != null)
doSignal(first);
}
doSignal
private void doSignal(Node first) {
do {
// 头节点指针往后移动一个节点
// 若当前头结点的后继节点为null，则尾节点置为null
if ( (firstWaiter = first.nextWaiter) == null)
lastWaiter = null;
// 将之前头节点的后继节点置为null，便于GC
first.nextWaiter = null;
// 头节点未成功转移到同步队列且头节点不为null，继续自旋
} while (!transferForSignal(first) &&
(first = firstWaiter) != null);
}
transferForSignal用于将条件等待队列节点转移到AQS的FIFO同步队列中。
final boolean transferForSignal(Node node) {
// 如果CAS失败，说明节点在signal之前被取消了，返回false
// 设置入队节点的等待状态为0
if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
return false;
// 设置成功则进行入队操作
Node p = enq(node);
// 获取入队节点的前继节点的状态
int ws = p.waitStatus;
// 1. 如果前继节点取消，则直接唤醒当前节点线程
// 2. 如果为非取消节点则将前继节点设置为SIGNAL，并不马上唤醒线程
if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
LockSupport.unpark(node.thread);
return true;
}
查询方法
isOwnedBy
// 判断当前Condition是否是由指定的AQS创建的
final boolean isOwnedBy(AbstractQueuedSynchronizer sync) {
return sync == AbstractQueuedSynchronizer.this;
}
hasWaiters
// 查询该Condition对象上是否有等待的线程
protected final boolean hasWaiters() {
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
for (Node w = firstWaiter; w != null; w = w.nextWaiter) {
if (w.waitStatus == Node.CONDITION)
return true;
}
return false;
}
getWaitQueueLength
// 获取条件等待队列的大约长度
// 因为没有在查询的时候加锁，并不能保证在查找的时候是否有新的节点加入
protected final int getWaitQueueLength() {
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
int n = 0;
for (Node w = firstWaiter; w != null; w = w.nextWaiter) {
if (w.waitStatus == Node.CONDITION)
++n;
}
return n;
}
getWaitingThreads
// 获取等待在Condition上的线程集合
protected final Collection<Thread> getWaitingThreads() {
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
ArrayList<Thread> list = new ArrayList<Thread>();
for (Node w = firstWaiter; w != null; w = w.nextWaiter) {
if (w.waitStatus == Node.CONDITION) {
Thread t = w.thread;
if (t != null)
list.add(t);
}
}
return list;
}
}
网友评论