Java并发编程之锁机制之Condition接口
前言
在前面的文章中，我曾提到过，整个Lock接口下实现的锁机制中AQS(AbstractQueuedSynchronizer，下文都称之为AQS)与Condition才是真正的实现者。也就说Condition在整个同步组件的基础框架中也起着很是重要的做用，既然它如此重要与犀利，那么如今咱们就一块儿去了解其内部的实际原理与具体逻辑。node
在阅读该文章以前，我由衷的建议先阅读《Java并发编程之锁机制之AQS》与《Java并发编程之锁机制之LockSupport工具》这两篇文章。由于整个Condtion的内部机制与逻辑都离不开以上两篇文章提到的知识点。编程
Condition接口方法介绍
在正式介绍Condtion以前，咱们能够先了解其中声明的方法。具体方法声明，以下表所示：数组
从该表中，咱们能够看出其内部定义了等待（以await开头系列方法）与通知（以singal开头的系列方法)两种类型的方法，相似于Object对象的wait()与notify()/NotifyAll()方法来对线程的阻塞与唤醒。bash
ConditionObject介绍
在实际使用中，Condition接口实现类是AQS中的内部类ConditionObject。在其内部维护了一个FIFO(first in first out)的队列（这里咱们称之为等待队列，你也能够叫作阻塞队列，看每一个人的理解），经过与AQS中的同步队列配合使用，来控制获取共享资源的线程。并发
等待队列
等待队列是ConditionObjec中内部的一个FIFO(first in first out)的队列，在队列中的每一个节点都包含了一个线程引用，且该线程就是在ConditionObject对象上阻塞的线程。须要注意的是，在等待队列中的节点是复用了AQS中Node类的定义。换句话说，在AQS中维护的同步队列与ConditionObjec中维护的等待队列中的节点类型都是AQS.Node类型。（关于AbstractQueuedSynchronizer.Node类的介绍，你们能够参看《Java并发编程之锁机制之AQS》文章中的描述）。框架
在ConditionObject类中也分别定义了firstWaiter与lastWaiter两个指针，分别指向等待队列中头部与尾部。当实际线程调用其以await开头的系列方法后。会将该线程构造为Node节点。添加等待队列中的尾部。关于等待队列的基本结构以下图所示：工具
对于等待队列中节点添加的方式也很简单，将上一尾节点的nextWaiter指向新添加的节点，同时使lastWaiter指向新添加的节点。post
同步队列与等待队列的对应关系
上文提到了整个Lock锁机制须要AQS中的同步队列与ConditionObject的等待队列配合使用，其对应关系以下图所示： ui
在Lock锁机制下，能够拥有一个同步队列和多个等待队列，与咱们传统的Object监视器模型上，一个对象拥有一个同步队列和等待队列不一样。lock中的锁能够伴有多个条件。this
Condition的基本使用
为了你们可以更好的理解同步队列与等待队列的关系。下面经过一个有界队列BoundedBuffer来了解Condition的使用方式，该类是一个特殊的队列，当队列为空时，队列的获取操做将会阻塞当前"拿"线程，直到队列中有新增的元素，当队列已满时，队列的放入操做将会阻塞"放入"线程，直到队列中出现空位。具体代码以下所示：
class BoundedBuffer {
final Lock lock = new ReentrantLock();
final Condition notFull = lock.newCondition();
final Condition notEmpty = lock.newCondition();
final Object[] items = new Object[100];
//依次为，放入的角标、拿的角标、数组中放入的对象总数
int putptr, takeptr, count;
/**
* 添加一个元素
* （1）若是当前数组已满，则把当前"放入"线程，加入到"放入"等待队列中，并阻塞当前线程
* （2）若是当前数组未满，则将x元素放入数组中，唤醒"拿"线程中的等待线程。
*/
public void put(Object x) throws InterruptedException {
lock.lock();
try {
while (count == items.length)//若是已满，则阻塞当前"放入"线程
notFull.await();
items[putptr] = x;
if (++putptr == items.length) putptr = 0;
++count;
notEmpty.signal();//唤醒"拿"线程
} finally {
lock.unlock();
}
}
/**
* 拿一个元素
* （1）若是当前数组已空，则把当前"拿"线程，加入到"拿"等待队列中，并阻塞当前线程
* （2）若是当前数组不为空，则把唤醒"放入"等待队列中的线程。
*/
public Object take() throws InterruptedException {
lock.lock();
try {
while (count == 0)//若是为空，则阻塞当前"拿"线程
notEmpty.await();
Object x = items[takeptr];
if (++takeptr == items.length) takeptr = 0;
--count;
notFull.signal();//唤醒"放入"线程
return x;
} finally {
lock.unlock();
}
}
}
复制代码
从代码中咱们能够看出，在该类中咱们建立了两个等待队列notFull与notEmpty。这两个等待队列的做用分别是，当请数组已满时，notFull用于存储阻塞的"放入"线程，notEmpty用于存储阻塞的"拿"线程。须要注意的是获取一个Condition必须经过Lock的newCondition()方法。关于ReentrantLock，在后续的文章中，咱们会进行介绍。
阻塞实现 await()
在了解了ConditionObject的内部基本结构和与AQS中内部的同步队列的对应关系后，如今咱们来看看其阻塞实现。调用ConditionObject的await()方法（或者以await开头的方法），会使当前线程进入等待队列，并释放同步状态，须要注意的是当该方法返回时，当前线程必定获取了同步状态（具体缘由是当经过signal()等系列方法，线程才会从await（）方法返回，而唤醒该线程后会加入同步队列）。这里咱们以awati()方法为例，具体代码以下所示：
public final void await() throws InterruptedException {
//若是当前线程已经中断，直接抛出异常
if (Thread.interrupted())
throw new InterruptedException();
//(1)将当前线程加入等待队列
Node node = addConditionWaiter();
//(2)释放同步状态（也就是释放锁），同时将线程节点从同步队列中移除，并唤醒同步队列中的下一节点
int savedState = fullyRelease(node);
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;
if (node.nextWaiter != null) // clean up if cancelled
unlinkCancelledWaiters();
if (interruptMode != 0)
reportInterruptAfterWait(interruptMode);
}
复制代码
从代码总体来看，整个方法分为如下四个步骤：
- （1）经过
addConditionWaiter()方法将线程节点加入到等待队列中。 - （2）经过
fullyRelease(Node node)方法释放同步状态（也就是释放锁），同时将线程节点从同步队列中移除，并唤醒同步队列中的下一节点。 - （3）经过
isOnSyncQueue(Node node)方法判断当前线程节点是否在同步队列中，若是不在，则经过LockSupport.park(this);阻塞当前线程。 - （4）当线程被唤醒后，调用
acquireQueued(node, savedState)方法，从新在同步队列中与其余线程竞争获取同步状态
由于每一个步骤涉及到的逻辑都稍微有一点复杂，这里为了方便你们理解，分别对以上四个步骤涉及到的方法分别进行介绍。
addConditionWaiter()方法
该方法主要将同步队列中的须要阻塞的线程节点加入到等待队列中，关于addConditionWaiter()方法具体代码以下所示：
private Node addConditionWaiter() {
Node t = lastWaiter;
// (1)若是当前尾节点中中对应的线程已经中断，
//则移除等待队列中全部的已经中断或已经释放同步状态的线程节点
if (t != null && t.waitStatus != Node.CONDITION) {
unlinkCancelledWaiters();
t = lastWaiter;
}
//(2)构建等待队列中的节点
Node node = new Node(Node.CONDITION);
//(3)将该线程节点添加到队列中，同时构建firstWaiter与lastWaiter的指向
if (t == null)
firstWaiter = node;
else
t.nextWaiter = node;
lastWaiter = node;
return node;
}
复制代码
该方法的逻辑也比较简单，分为如下三个步骤：
- （1）获取等待队列中的尾节点，若是当前尾节点已经中断，那么则经过
unlinkCancelledWaiters()方法移除等待队列中全部的已经中断或已经释放同步状态（也就是释放锁）的线程节点 - （2）构建等待队列中的节点，注意，是经过
New的形式，那么就说明与同步队列中的线程节点不是同一个。（对Node状态枚举不清楚的小伙伴，能够参看Java并发编程之锁机制之AQS文章下的Node状态枚举介绍）。 - （3）将该线程节点添加到等待队列中去，同时构建firstWaiter与lastWaiter的指向，能够看出等待队列老是以
FIFO(first in first out )的形式添加线程节点。
unlinkCancelledWaiters()方法
由于在addConditionWaiter()方法的步骤（1）中，调用了unlinkCancelledWaiters移除了全部的已经中断的线程节点，那咱们看一个该方法的实现。以下所示：
private void unlinkCancelledWaiters() {
//获取等待队列中的头节点
Node t = firstWaiter;
Node trail = null;
//遍历等待队列，将已经中断的线程节点从等待队列中移除。
while (t != null) {
Node next = t.nextWaiter;
if (t.waitStatus != Node.CONDITION) {
t.nextWaiter = null;
if (trail == null)
firstWaiter = next;
else
trail.nextWaiter = next;
if (next == null)//从新定义lastWaiter的指向
lastWaiter = trail;
}
else
trail = t;
t = next;
}
}
复制代码
该方法具体流程以下图所示：
fullyRelease(Node node)
在将阻塞线程将入到等待队列后，会将该线程节点从同步队列中移除，释放同步状态（也就是释放锁），并唤醒同步队列中的下一节点。具体代码以下所示：
final int fullyRelease(Node node) {
try {
int savedState = getState();
if (release(savedState))
return savedState;
throw new IllegalMonitorStateException();
} catch (Throwable t) {
node.waitStatus = Node.CANCELLED;
throw t;
}
}
复制代码
release(int arg)方法会释放当前线程的同步状态， 并唤醒同步队列中的下一线程节点，使其尝试获取同步状态，由于该方法已经在Java并发编程之锁机制之AQS文章下的unparkSuccessorNode node)方法的下分析过了，因此这里就再也不进行分析了。但愿你们参考上面提到的文章进行理解。
isOnSyncQueue(Node node)
该方法主要用于判断当前线程节点是否在同步队列中。具体代码以下所示：
final boolean isOnSyncQueue(Node node) {
//判断当前节点 waitStatus ==Node.CONDITION或者当前节点上一节点为空,则不在同步队列中
if (node.waitStatus == Node.CONDITION || node.prev == null)
return false;
//若是当前节点拥有下一个节点，则在同步队列中。
if (node.next != null) // If has successor, it must be on queue
return true;
//若是以上条件都不知足，则遍历同步队列。检查是否在同步队列中。
return findNodeFromTail(node);
}
复制代码
若是你还记得AQS中的同步队列，那么你应该知道同步队列中的Node节点才会使用其内部的pre与next字段，那么在同步队列中由于只使用了nextWaiter字段，因此咱们就能很简单的经过这两个字段是否为==null，来判断是否在同步队列中。固然也有可能有一种特殊状况。有可能须要阻塞的线程节点尚未加入到同步队列中，那么这个时候咱们须要遍历同步队列来判断是否在该线程节点是否在线程中。具体代码以下所示：
private boolean findNodeFromTail(Node node) {
for (Node p = tail;;) {
if (p == node)
return true;
if (p == null)
return false;
p = p.prev;
}
}
复制代码
这里之因此使用同步队列tail（尾节点）来遍历，若是node.netx!=null，那么就说明当前线程已经在同步队列中。那么咱们须要处理的状况确定是针对node.next==null的状况。因此须要从尾节点开始遍历。
acquireQueued(final Node node, int arg)
当线程被唤醒后（具体缘由是当经过signal()等系列方法，线程才会从await（）方法返回）会调用该方法将该线程节点加入到同步队列中。该方法我在《Java并发编程之锁机制之AQS》中具体描述过了。这里就不在进行过多的解析。
阻塞流程
在理解了整个阻塞的流程后，如今咱们来概括总结一下，整个阻塞的流程。具体流程以下图所示：
- （1）将该线程节点从同步队列中移除，并释放其同步状态。
- （2）构造新的阻塞节点，加入到等待队列中。
唤醒实现 signal()
当须要唤醒线程时，会调用ConditionObject中的singal开头的系列方法，该系列方法会唤醒等待队列中的首个线程节点，在唤醒该节点以前，会先讲该节点移动到同步队列中。这里咱们以singal()方法为例进行讲解，具体代码以下：
public final void signal() {
//（1）判断当前线程是否获取到了同步状态（也就是锁）
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
Node first = firstWaiter;
//（2）获取等待队列中的首节点，而后将其移动到同步队列，而后再唤醒该线程节点
if (first != null)
doSignal(first);
}
复制代码
该方法主要逻辑分为如下两个步骤：
- （1）经过
isHeldExclusively()方法，判断当前线程是否获取到了同步状态（也就是锁）。 - （2）经过
doSignal(Node first)方法，获取等待队列中的首节点，而后将其移动到同步队列，而后再唤醒该线程节点。
下面咱们会分别对上面涉及到的两个方法进行描述。
isHeldExclusively()方法
isHeldExclusively()方法是AQS中的方法，默认交给其子类实现，主要用于判断当前调用singal()方法的线程，是否在同步队列中，且已经获取了同步状态。具体代码以下所示：
protected boolean isHeldExclusively() {
throw new UnsupportedOperationException();
}
复制代码
doSignal(Node first)方法
那咱们继续跟踪doSignal(Node first)方法，具体方法以下：
private void doSignal(Node first) {
do {
//（1）将等待队列中的首节点从等待队列中移除，并从新制定firstWaiter的指向
if ( (firstWaiter = first.nextWaiter) == null)
lastWaiter = null;
first.nextWaiter = null;
} while (!transferForSignal(first) &&
//（2）将等待队列中的首节点，加入同步队列中，并从新唤醒该节点
(first = firstWaiter) != null);
}
复制代码
该方法也很简单，分为两个步骤：
- （1）将等待队列中的首节点从等待队列中移除，并设置firstWaiter的指向为首节点的下一个节点。 为了方便你们理解该步骤所描述的逻辑，这里画了具体的图，具体状况以下图所示：
- （2）经过
transferForSignal(Node node)方法，将等待队列中的首节点，加入到同步队列中去，而后从新唤醒该线程节点。
transferForSignal(Node node)方法
由于步骤（2）中transferForSignal(Node node)方法较为复杂，因此会对该方法进行详细的讲解。具体代码以下所示：
final boolean transferForSignal(Node node) {
//（1）将该线程节点的状态设置为初始状态，若是失败则表示当前线程已经中断了
if (!node.compareAndSetWaitStatus(Node.CONDITION, 0))
return false;
//（2）将该节点放入同步队列中，
Node p = enq(node);
int ws = p.waitStatus;
//（3）获取当前节点的状态并判断，尝试将该线程节点状态设置为Singal，若是失败则唤醒线程
if (ws > 0 || !p.compareAndSetWaitStatus(ws, Node.SIGNAL))
LockSupport.unpark(node.thread);
return true;
}
复制代码
该方法分为三个步骤：
- （1）将该线程节点的状态设置为初始状态，若是失败则表示当前线程已经中断了，直接返回。
- （2）经过
enq(Node node)方法，将该线程节点放入同步队列中。 - （3）当将该线程节点放入同步队列后，获取当前节点的状态并判断，若是该节点的
waitStatus>0或者经过compareAndSetWaitStatus(ws, Node.SIGNAL)将该节点的状态设置为Singal，若是失败则经过LockSupport.unpark(node.thread)唤醒线程。
上述步骤中，着重讲enq(Node node)方法，关于LockSupport.unPark(Thread thread)方法的理解，你们能够阅读《Java并发编程之锁机制之LockSupport工具》。下面咱们就来分析enq(Node node)方法。具体代码以下所示：
private Node enq(Node node) {
for (;;) {
//(1)获取同步队列的尾节点
Node oldTail = tail;
//(2)若是尾节点不为空，则将该线程节点加入到同步队列中
if (oldTail != null) {
//将当前节点的prev指向尾节点
U.putObject(node, Node.PREV, oldTail);
//将同步队列中的tail指针，指向当前节点
if (compareAndSetTail(oldTail, node)) {
oldTail.next = node;
return oldTail;
}
} else {
//(3)若是当前同步队列为空，则构造同步队列
initializeSyncQueue();
}
}
}
复制代码
观察该方法，咱们发现该方法经过死循环(固然你也能够叫作自旋)的方式来添加该节点到同步队列中去。该方法分为如下步骤：
- （1）获取同步队列的尾节点
- （2）若是尾节点不为空，则将该线程节点加入到同步队列中
- （3）若是当前同步队列为空，则经过
initializeSyncQueue();构造同步队列。
这里对Node enq(Node node)中的步骤（2）补充一个知识点。咱们来看一下调用U.putObject(node, Node.PREV, oldTail);语句，内部是如何将当前的节点的prev指向尾节点的。在AQS（AbstractQueuedSynchronizer）中的Node类中有以下静态变量和语句。这里我省略了一下不重要的代码。具体代码以下所示：
private static final sun.misc.Unsafe U = sun.misc.Unsafe.getUnsafe();
//省略部分代码
static final long PREV;
static {
try {
//省略部分代码
PREV = U.objectFieldOffset
(Node.class.getDeclaredField("prev"));
} catch (ReflectiveOperationException e) {
throw new Error(e);
}
}
}
复制代码
其中Node.class.getDeclaredField("prev")语句很好理解，就是获取Node类中pre字段，若是有则返回相应Field字段，反之抛出NoSuchFieldException异常。关于Unfase中的objectFieldOffset(Field f)方法，我曾经在《Java并发编程之锁机制之LockSupport工具》描述过相似的状况。这里我简单的再解释一遍。该方法用于获取某个字段相对 Java对象的“起始地址”的偏移量，也就是说每一个字段在类对应的内存中存储是有“角标”的，那么也就是说咱们如今的PREV静态变量就表明着Node中prev字段在内存中的“角标”。
当获取到"角标"后，咱们再经过U.putObject(node, Node.PREV, oldTail);该方法第一个参数是操做对象，第二个参数是操做的内存“角标”，第三个参数是指望值。那么最后，也就完成了将当前节点的prev字段指向同步队列的尾节点。
当理解了该知识点后，剩下的将同步队列中的tail指针，指向当前节点与若是当前同步队列为空，则构造同步队列这两个操做就很是好理解了。因为篇幅的限制，在这里我就不在进行描述了。但愿读者朋友们，能阅读源代码，触类旁通。关于这两个方法的代码以下所示：
private static final sun.misc.Unsafe U = sun.misc.Unsafe.getUnsafe();
private static final long STATE;
private static final long HEAD;
private static final long TAIL;
static {
try {
STATE = U.objectFieldOffset
(AbstractQueuedSynchronizer.class.getDeclaredField("state"));
HEAD = U.objectFieldOffset
(AbstractQueuedSynchronizer.class.getDeclaredField("head"));
TAIL = U.objectFieldOffset
(AbstractQueuedSynchronizer.class.getDeclaredField("tail"));
} catch (ReflectiveOperationException e) {
throw new Error(e);
}
Class<?> ensureLoaded = LockSupport.class;
}
private final void initializeSyncQueue() {
Node h;
if (U.compareAndSwapObject(this, HEAD, null, (h = new Node())))
tail = h;
}
private final boolean compareAndSetTail(Node expect, Node update) {
return U.compareAndSwapObject(this, TAIL, expect, update);
}
复制代码
唤醒流程
在理解了唤醒的具体逻辑后，如今来总结一下，唤醒的具体流程。具体以下图所示：
- 将等待队列中的
头节点线程，移动到同步队列中。 - 当移动到同步队列中后。唤醒该线程。是该线程参与同步状态的竞争。
总体流程其实不算太复杂，你们只须要注意，当咱们将等待队列中的线程节点加入到同步队列以后，才会唤醒线程。
最后
该文章参考如下图书，站在巨人的肩膀上。能够看得更远。
- 《Java并发编程的艺术》
推荐阅读
- 1. Java 并发编程之 Condition 接口
- 2. Java并发编程之锁机制之Lock接口
- 3. JAVA并发编程（13）-Condition接口
- 4. Java并发编程之锁机制之（ReentrantLock)重入锁
- 5. Java并发编程之锁机制之ReentrantReadWriteLock（读写锁）
- 6. Java并发--Condition接口
- 7. Java 并发？Condition接口
- 8. Java并发编程J.U.C之Condition
- 9. 并发编程之java锁
- 10. 并发编程之Java锁
- 更多相关文章...
- • Rust 并发编程 - RUST 教程
- • Eclipse 创建 Java 接口 - Eclipse 教程
- • 漫谈MySQL的锁机制
- • Java 8 Stream 教程
-
每一个你不满意的现在，都有一个你没有努力的曾经。
- 1. CVPR 2020 论文大盘点-光流篇
- 2. Photoshop教程_ps中怎么载入图案？PS图案如何导入？
- 3. org.pentaho.di.core.exception.KettleDatabaseException:Error occurred while trying to connect to the
- 4. SonarQube Scanner execution execution Error --- Failed to upload report - 500: An error has occurred
- 5. idea 导入源码包
- 6. python学习 day2——基础学习
- 7. 3D将是页游市场新赛道？
- 8. osg--交互
- 9. OSG-交互
- 10. Idea、spring boot 图片(pgn显示、jpg不显示)解决方案