}
}
public E take() throws InterruptedException {
lock.lock();
try {
// 当数组为空时，调用notEmpty.await();使得获取元素的线程阻塞
while (count =
《一线大厂Java面试题解析+后端开发学习笔记+最新架构讲解视频+实战项目源码讲义》
【docs.qq.com/doc/DSmxTbFJ1cmN1R2dB】 完整内容开源分享
= 0) {
notEmpty.await();
}
E ret = items[head];
items[head] = null;
if (++head == items.length) {
head = 0;
}
–count;
// 唤醒插入元素的线程
notFull.signalAll();
return ret;
} finally {
lock.unlock();
}
}
}
上述案例中的tail、head、count有不大看的懂的，我做了个简单的的put(E e)和take()的图解
Condition实现的阻塞队列put(E e)图解
Condition实现的阻塞队列take()图解
4、Condtion 实现源码分析 4.1 互斥锁和读写锁中Condition的构造 4.1.1 AbstractQueuedSynchronizer中的ConditionObjectReentrantLock与ReentrantReadWriteLock中的静态内部类Sync继承了AbstractQueuedSynchronizer，两者调用的sync.newCondition()，实际上调用的是new ConditionObject()，也就是构造的AbstractQueuedSynchronizer中的ConditionObject对象。
public class ConditionObject implements Condition, java.io.Serializable {
private transient Node firstWaiter;
private transient Node lastWaiter;
public ConditionObject() { }
//…
}
4.1.2 ReentrantLock中的Conditionpublic Condition newCondition() {
return sync.newCondition();
}
final ConditionObject newCondition() {
return new ConditionObject();
}
4.1.3 ReentrantReadWriteLock中的Conditionpublic static class ReadLock implements Lock, java.io.Serializable {
public Condition newCondition() {
throw new UnsupportedOperationException();
}
}
public static class WriteLock implements Lock, java.io.Serializable {
public Condition newCondition() {
return sync.newCondition();
}
}
4.2 await()源码分析 4.2.1 await()位于AbstractQueuedSynchronizer中的ConditionObject调用Condition的await()或者awaitXxxx()会导致线程构建成Node节点加入Condition的等待队列，并且释放锁。如果线程从await()或者awaitXxxI()方法返回，表明线程又重新获取了Condition相关的锁。
public final void await() throws InterruptedException {
// 判断当前线程是否被中断，如果被中断了则抛出InterruptedException异常
if (Thread.interrupted())
throw new InterruptedException();
// 添加一个新的等待节点到等待队列，无需CAS因为调用await前提是获取到了锁
Node node = addConditionWaiter();
// 释放锁，调用await()必须获取锁，此处释放锁可以防止死锁产生
int savedState = fullyRelease(node);
int interruptMode = 0;
// isonSyncQueue(node)判断当前节点是否加入到同步队列中了，也就是移出了等待队列+ `SWDVVB
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)
unlinkCancelledWaiters();
// 响应中断抛出异常
if (interruptMode != 0)
reportInterruptAfterWait(interruptMode);
}
await()中的addConditionWaiter()方法
private Node addConditionWaiter() {
Node t = lastWaiter;
// 清除取消等待的节点
if (t != null && t.waitStatus != Node.CONDITION) {
unlinkCancelledWaiters();
t = lastWaiter;
}
// 构造新的节点Node
Node node = new Node(Thread.currentThread(), Node.CONDITION);
// 如果尾节点为空免责证明首节点也为空，firstWaiter = node;
if (t == null)
firstWaiter = node;
else
t.nextWaiter = node; // 尾节点的下一个节点指向当前节点，此时当前节点成为新的尾节点
lastWaiter = node;
return node;
}
4.2.2 总结-
await()方法在调用之前，线程一定获取到了锁，因此addConditionWaiter()无需CAS也可以保证线程安全
-
在阻塞自己之前，必须先释放锁fullyRelease(node)，防止死锁
-
线程从wait中被唤醒后，必须通过acquireQueued(node, savedState)重新获取锁
-
isonSyncQueue(node)用于判断节点是否在AQS同步队列中（关于同步队列和等待队列文章后面有图解），如果从Condition的等待队列移动到了AQS的同步队列证明被执行了signal()
-
LockSupport.park(this)阻塞自己之后，线程被唤醒的方式有unpark和中断，通过checkInterruptWhileWaiting(node)判断当前线程被唤醒是否是因为中断，如果中断则退出循环
调用Condition的signal()方法，会唤醒在Condition等待队列中的线程节点（唤醒的是等待时间最长的首节点），唤醒节点之前会将其移至同步队列中（这里要注意先加入同步队列在唤醒该节点，等会画图别混淆）。
public final void signal() {
// 判断当前线程是否获取了锁，如果没有则抛出异常
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
// 构建获取等待队列的首节点
Node first = firstWaiter;
// 如果首节点不为空
if (first != null)
// 在下面
doSignal(first);
}
signal()中调用多的doSignal(Node first)方法
private void doSignal(Node first) {
do {
// 判断当前结束是否是最后一个等待队列中的节点
if ( (firstWaiter = first.nextWaiter) == null)
lastWaiter = null;
first.nextWaiter = null;
// 在下面
} while (!transferForSignal(first) &&
(first = firstWaiter) != null);
}
doSignal(Node first)中调用的transferForSignal(Node node)方法
final boolean transferForSignal(Node node) {
if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
return false;
// enq将当前节点加入AQS的同步队列，这个我在前面的文中讲过
Node p = enq(node);
int ws = p.waitStatus;
if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
// 唤醒阻塞中的线程，先加入同步队列，再唤醒
LockSupport.unpark(node.thread);
return true;
}
5、Condition实现原理图解 5.1 图解同步队列与等待队列在文章开头第二大点介绍Condition之于Lock与wati()/notify()之于synchronized时，我们对比过二者，其中很大的一个区别在于Object的监视模型上，一个对象只拥有一个同步队列和等待队列，这样的模型一个很大的问题在于它不太适用于编写带有多个条件谓词的并发对象（可以简单理解为复杂的带高级功能的）；而并发包中的Lock中的组合了Condition对象，使得其可以拥有一个同步队列和多个等待队列（一个Condition中有一个等待队列）。下面就通过图来说明Condition的等待队列和同步器中的同步队列和等待队列之间的关系。
Condition等待队列图示
AQS同步队列与Condition等待队列关系图示
5.2 图解await()方法如何加入等待队列5.3 图解signal()方法如何移出等待队列前面讲过，调用await()方法的前提是获取到了Lock对应的锁，也正是因为这个await()操作是在获取锁的前提下进行的，所以节点的构造并未使用CAS，因为它的前提条件就是线程互斥（安全）的；同时我们在讲述AQS、ReentrantLock和ReentrantReadWriteLock时讲述过其线程竞争锁资源失败，线程将会被构造成同步节点，加入AQS的同步队列中，等待后续的再次竞争或者中断退出等。上图也讲过了同步器AQS中的同步队列和Condition中的等待队列之间的关系，所以加入Condition等待队列的线程，可以理解为在AQS同步器中重新获取到锁的首节点线程被移植（这里的移植不是将以前的节点加入，是通过以前节点的信息构造一个新的线程节点加入到等待队列）到了Condition的等待队列中，其图如下。
6、总结signal()方法会将等待队列中的等待时间最长的节点（首节点），移动到同步队列尾部，加入同步队列的代码是 enq(final Node node)，通过CAS来线程安全的移动。移动完成之后线程再使用LockSupport.unpark(node.thread);唤醒该节点中等待的线程。线程节点被移动至同步队列中后，线程可以参与同步状态的竞争，如果竞争成功，线程将会从await()方法返回。其图解如下。
Condition中的await()/signal()相比Object中的wait()/notify()，Condition拥有更多的高级特性能够实现更加复杂的等待线程集的场景。但是我们在使用Lock和Condition时在调用await()和signal()要注意必须持有Lock对象（尽管在Lock对象中定义的具体实现，构造一个Condition可以不满足持有Lock对象这个条件）。
