Condition源码分析与等待通知机制，linux系统架构与运维实战pdf

Condition源码分析与等待通知机制，linux系统架构与运维实战pdf

//后继节点

Node nextWaiter;

进一步说明，等待队列是一个单向队列，而在之前说AQS时知道同步队列是一个双向队列。接下来我们用一个demo，通过debug进去看是不是符合我们的猜想：

public static void main(String[] args) {

for (int i = 0; i < 10; i++) {

Thread thread = new Thread(() -> {

lock.lock();

【一线大厂Java面试题解析+后端开发学习笔记+最新架构讲解视频+实战项目源码讲义】

浏览器打开：qq.cn.hn/FTf 免费领取

try {

condition.await();

} catch (InterruptedException e) {

e.printStackTrace();

}finally {

lock.unlock();

}

});

thread.start();

}

}

这段代码没有任何实际意义，甚至很臭，只是想说明下我们刚才所想的。新建了10个线程，没有线程先获取锁，然后调用condition.await方法释放锁将当前线程加入到等待队列中，通过debug控制当走到第10个线程的时候查看firstWaiter即等待队列中的头结点，debug模式下情景图如下：

从这个图我们可以很清楚的看到这样几点：1. 调用condition.await方法后线程依次尾插入到等待队列中，如图队列中的线程引用依次为Thread-0,Thread-1,Thread-2…Thread-8；2. 等待队列是一个单向队列。通过我们的猜想然后进行实验验证，我们可以得出等待队列的示意图如下图所示：

同时还有一点需要注意的是：我们可以多次调用lock.newCondition()方法创建多个condition对象，也就是一个lock可以持有多个等待队列。而在之前利用Object的方式实际上是指在对象Object对象监视器上只能拥有一个同步队列和一个等待队列，而并发包中的Lock拥有一个同步队列和多个等待队列。示意图如下：

如图所示，ConditionObject是AQS的内部类，因此每个ConditionObject能够访问到AQS提供的方法，相当于每个Condition都拥有所属同步器的引用。

[](

)await实现原理

当调用condition.await()方法后会使得当前获取lock的线程进入到等待队列，如果该线程能够从await()方法返回的话一定是该线程获取了与condition相关联的lock。接下来，我们还是从源码的角度去看，只有熟悉了源码的逻辑我们的理解才是最深的。await()方法源码为：

public final void await() throws InterruptedException {

if (Thread.interrupted())

throw new InterruptedException();

// 1. 将当前线程包装成Node，尾插入到等待队列中

Node node = addConditionWaiter();

// 2. 释放当前线程所占用的lock，在释放的过程中会唤醒同步队列中的下一个节点

int savedState = fullyRelease(node);

int interruptMode = 0;

while (!isonSyncQueue(node)) {

// 3. 当前线程进入到等待状态

LockSupport.park(this);

if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)

break;

}

// 4. 自旋等待获取到同步状态（即获取到lock）

if (acquireQueued(node, savedState) && interruptMode != THROW_IE)

interruptMode = REINTERRUPT;

if (node.nextWaiter != null) // clean up if cancelled

unlinkCancelledWaiters();

// 5. 处理被中断的情况

if (interruptMode != 0)

reportInterruptAfterWait(interruptMode);

}

代码的主要逻辑请看注释，我们都知道当前线程调用condition.await()方法后，会使得当前线程释放lock然后加入到等待队列中，直至被signal/signalAll后会使得当前线程从等待队列中移至到同步队列中去，直到获得了lock后才会从await方法返回，或者在等待时被中断会做中断处理。那么关于这个实现过程我们会有这样几个问题：1. 是怎样将当前线程添加到等待队列中去的？2.释放锁的过程？3.怎样才能从await方法退出？而这段代码的逻辑就是告诉我们这三个问题的答案。具体请看注释，在第1步中调用addConditionWaiter将当前线程添加到等待队列中，该方法源码为：

private Node addConditionWaiter() {

Node t = lastWaiter;

// If lastWaiter is cancelled, clean out.

if (t != null && t.waitStatus != Node.CONDITION) {

unlinkCancelledWaiters();

t = lastWaiter;

}

//将当前线程包装成Node

Node node = new Node(Thread.currentThread(), Node.CONDITION);

if (t == null)

firstWaiter = node;

else

//尾插入

t.nextWaiter = node;

//更新lastWaiter

lastWaiter = node;

return node;

}

这段代码就很容易理解了，将当前节点包装成Node，如果等待队列的firstWaiter为null的话（等待队列为空队列），则将firstWaiter指向当前的Node，否则，更新lastWaiter(尾节点)即可。就是通过尾插入的方式将当前线程封装的Node插入到等待队列中即可，同时可以看出等待队列是一个不带头结点的链式队列，之前我们学习AQS时知道同步队列是一个带头结点的链式队列，这是两者的一个区别。将当前节点插入到等待对列之后，会使当前线程释放lock，由fullyRelease方法实现，fullyRelease源码为：

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;

}

}

这段代码就很容易理解了，调用AQS的模板方法release方法释放AQS的同步状态并且唤醒在同步队列中头结点的后继节点引用的线程，如果释放成功则正常返回，若失败的话就抛出异常。到目前为止，这两段代码已经解决了前面的两个问题的答案了，还剩下第三个问题，怎样从await方法退出？现在回过头再来看await方法有这样一段逻辑：

while (!isonSyncQueue(node)) {

// 3. 当前线程进入到等待状态

LockSupport.park(this);

if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)

break;

}

很显然，当线程第一次调用condition.await()方法时，会进入到这个while()循环中，然后通过LockSupport.park(this)方法使得当前线程进入等待状态，那么要想退出这个await方法第一个前提条件自然而然的是要先退出这个while循环，出口就只剩下两个地方：1. 逻辑走到break退出while循环；2. while循环中的逻辑判断为false。再看代码出现第1种情况的条件是当前等待的线程被中断后代码会走到break退出，第二种情况是当前节点被移动到了同步队列中（即另外线程调用的condition的signal或者signalAll方法），while中逻辑判断为false后结束while循环。总结下，就是当前线程被中断或者调用condition.signal/condition.signalAll方法，当前节点移动到了同步队列后 ，这是当前线程退出await方法的前提条件。当退出while循环后就会调用acquireQueued(node, savedState)，这个方法在介绍AQS的底层实现时说过了，若感兴趣的话可以去[看这篇文章](

)，该方法的作用是在自旋过程中线程不断尝试获取同步状态，直至成功（线程获取到lock）。这样也说明了退出await方法必须是已经获得了condition引用（关联）的lock。到目前为止，开头的三个问题我们通过阅读源码的方式已经完全找到了答案，也对await方法的理解加深。await方法示意图如下图：

如图，调用condition.await方法的线程必须是已经获得了lock，也就是当前线程是同步队列中的头结点。调用该方法后会使得当前线程所封装的Node尾插入到等待队列中。

超时机制的支持

condition还额外支持了超时机制，使用者可调用方法awaitNanos,awaitUtil。这两个方法的实现原理，基本上与AQS中的tryAcquire方法如出一辙，关于tryAcquire可以仔细阅读[这篇文章](

)。

不响应中断的支持

要想不响应中断可以调用condition.awaitUninterruptibly()方法，该方法的源码为：

public final void awaitUninterruptibly() {

Node node = addConditionWaiter();

int savedState = fullyRelease(node);

boolean interrupted = false;

while (!isonSyncQueue(node)) {

LockSupport.park(this);

if (Thread.interrupted())

interrupted = true;

}

if (acquireQueued(node, savedState) || interrupted)

selfInterrupt();

}

这段方法与上面的await方法基本一致，只不过减少了对中断的处理，并省略了reportInterruptAfterWait方法抛被中断的异常。

[](

)signal/signalAll实现原理

调用condition的signal或者signalAll方法可以将等待队列中等待时间最长的节点移动到同步队列中，使得该节点能够有机会获得lock。按照等待队列是先进先出（FIFO）的，所以等待队列的头节点必然会是等待时间最长的节点，也就是每次调用condition的signal方法是将头节点移动到同步队列中。我们来通过看源码的方式来看这样的猜想是不是对的，signal方法源码为：

public final void signal() {

//1. 先检测当前线程是否已经获取lock

if (!isHeldExclusively())

throw new IllegalMonitorStateException();

//2. 获取等待队列中第一个节点，之后的 *** 作都是针对这个节点

Node first = firstWaiter;

if (first != null)

doSignal(first);

}

signal方法首先会检测当前线程是否已经获取lock，如果没有获取lock会直接抛出异常，如果获取的话再得到等待队列的头指针引用的节点，之后的 *** 作的doSignal方法也是基于该节点。下面我们来看看doSignal方法做了些什么事情，doSignal方法源码为：

private void doSignal(Node first) {

do {

if ( (firstWaiter = first.nextWaiter) == null)

lastWaiter = null;

//1. 将头结点从等待队列中移除

first.nextWaiter = null;

//2. while中transferForSignal方法对头结点做真正的处理

} while (!transferForSignal(first) &&

(first = firstWaiter) != null);

}

具体逻辑请看注释，真正对头节点做处理的逻辑在transferForSignal放，该方法源码为：

final boolean transferForSignal(Node node) {

//1. 更新状态为0

if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))

return false;

//2.将该节点移入到同步队列中去

Node p = enq(node);