ReentrantLock源码详解--条件锁

问题

（1）条件锁是什么？

（2）条件锁适用于什么场景？

（3）条件锁的await()是在其它线程signal()的时候唤醒的吗？

简介

条件锁，是指在获取锁之后发现当前业务场景自己无法处理，而需要等待某个条件的出现才可以继续处理时使用的一种锁。

比如，在阻塞队列中，当队列中没有元素的时候是无法弹出一个元素的，这时候就需要阻塞在条件notempty上，等待其它线程往里面放入一个元素后，唤醒这个条件notempty，当前线程才可以继续去做“弹出一个元素”的行为。

注意，这里的条件，必须是在获取锁之后去等待，对应到reentrantlock的条件锁，就是获取锁之后才能调用condition.await()方法。

在java中，条件锁的实现都在aqs的conditionobject类中，conditionobject实现了condition接口，下面我们通过一个例子来进入到条件锁的学习中。

使用示例

public class reentrantlocktest {
public static void main(string[] args) throws interruptedexception {
// 声明一个重入锁
reentrantlock lock = new reentrantlock();
// 声明一个条件锁
condition condition = lock.newcondition();
new thread(()->{
try {
lock.lock(); // 1
try {
system.out.println("before await"); // 2
// 等待条件
condition.await(); // 3
system.out.println("after await"); // 10
} finally {
lock.unlock(); // 11
}
} catch (interruptedexception e) {
e.printstacktrace();
}
}).start();
// 这里睡1000ms是为了让上面的线程先获取到锁
thread.sleep(1000);
lock.lock(); // 4
try {
// 这里睡2000ms代表这个线程执行业务需要的时间
thread.sleep(2000); // 5
system.out.println("before signal"); // 6
// 通知条件已成立
condition.signal(); // 7
system.out.println("after signal"); // 8
} finally {
lock.unlock(); // 9
}
}
}

上面的代码很简单，一个线程等待条件，另一个线程通知条件已成立，后面的数字代表代码实际运行的顺序，如果你能把这个顺序看懂基本条件锁掌握得差不多了。

源码分析

conditionobject的主要属性

public class conditionobject implements condition, java.io.serializable {
/** first node of condition queue. */
private transient node firstwaiter;
/** last node of condition queue. */
private transient node lastwaiter;
}

可以看到条件锁中也维护了一个队列，为了和aqs的队列区分，我这里称为条件队列，firstwaiter是队列的头节点，lastwaiter是队列的尾节点，它们是干什么的呢？接着看。

lock.newcondition()方法

新建一个条件锁。

// reentrantlock.newcondition()
public condition newcondition() {
return sync.newcondition();
}
// reentrantlock.sync.newcondition()
final conditionobject newcondition() {
return new conditionobject();
}
// abstractqueuedsynchronizer.conditionobject.conditionobject()
public conditionobject() { }

新建一个条件锁最后就是调用的aqs中的conditionobject类来实例化条件锁。

condition.await()方法

condition.await()方法，表明现在要等待条件的出现。

// abstractqueuedsynchronizer.conditionobject.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)) {
// 阻塞当前线程
locksupport.park(this);

// 上面部分是调用await()时释放自己占有的锁，并阻塞自己等待条件的出现
// *************************分界线************************* //
// 下面部分是条件已经出现，尝试去获取锁

if ((interruptmode = checkinterruptwhilewaiting(node)) != 0)
break;
}

// 尝试获取锁，注意第二个参数，这是上一章分析过的方法
// 如果没获取到会再次阻塞（这个方法这里就不贴出来了，有兴趣的翻翻上一章的内容）
if (acquirequeued(node, savedstate) && interruptmode != throw_ie)
interruptmode = reinterrupt;
// 清除取消的节点
if (node.nextwaiter != null) // clean up if cancelled
unlinkcancelledwaiters();
// 线程中断相关
if (interruptmode != 0)
reportinterruptafterwait(interruptmode);
}
// abstractqueuedsynchronizer.conditionobject.addconditionwaiter
private node addconditionwaiter() {
node t = lastwaiter;
// 如果条件队列的尾节点已取消，从头节点开始清除所有已取消的节点
if (t != null && t.waitstatus != node.condition) {
unlinkcancelledwaiters();
// 重新获取尾节点
t = lastwaiter;
}
// 新建一个节点，它的等待状态是condition
node node = new node(thread.currentthread(), node.condition);
// 如果尾节点为空，则把新节点赋值给头节点（相当于初始化队列）
// 否则把新节点赋值给尾节点的nextwaiter指针
if (t == null)
firstwaiter = node;
else
t.nextwaiter = node;
// 尾节点指向新节点
lastwaiter = node;
// 返回新节点
return node;
}
// abstractqueuedsynchronizer.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;
}
}
// abstractqueuedsynchronizer.isonsyncqueue
final boolean isonsyncqueue(node node) {
// 如果等待状态是condition，或者前一个指针为空，返回false
// 说明还没有移到aqs的队列中
if (node.waitstatus == node.condition || node.prev == null)
return false;
// 如果next指针有值，说明已经移到aqs的队列中了
if (node.next != null) // if has successor, it must be on queue
return true;
// 从aqs的尾节点开始往前寻找看是否可以找到当前节点，找到了也说明已经在aqs的队列中了
return findnodefromtail(node);
}

这里有几个难理解的点：

（1）condition的队列和aqs的队列不完全一样；

aqs的队列头节点是不存在任何值的，是一个虚节点；

condition的队列头节点是存储着实实在在的元素值的，是真实节点。

（2）各种等待状态（waitstatus）的变化；

首先，在条件队列中，新建节点的初始等待状态是condition（-2）；

其次，移到aqs的队列中时等待状态会更改为0（aqs队列节点的初始等待状态为0）；

然后，在aqs的队列中如果需要阻塞，会把它上一个节点的等待状态设置为signal（-1）；

最后，不管在condition队列还是aqs队列中，已取消的节点的等待状态都会设置为cancelled（1）；

另外，后面我们在共享锁的时候还会讲到另外一种等待状态叫propagate（-3）。

（3）相似的名称；

aqs中下一个节点是next，上一个节点是prev；

condition中下一个节点是nextwaiter，没有上一个节点。

如果弄明白了这几个点，看懂上面的代码还是轻松加愉快的，如果没弄明白，彤哥这里指出来了，希望您回头再看看上面的代码。

下面总结一下await()方法的大致流程：

（1）新建一个节点加入到条件队列中去；

（2）完全释放当前线程占有的锁；

（3）阻塞当前线程，并等待条件的出现；

（4）条件已出现（此时节点已经移到aqs的队列中），尝试获取锁；

也就是说await()方法内部其实是先释放锁->等待条件->再次获取锁的过程。

condition.signal()方法

condition.signal()方法通知条件已经出现。

// abstractqueuedsynchronizer.conditionobject.signal
public final void signal() {
// 如果不是当前线程占有着锁，调用这个方法抛出异常
// 说明signal()也要在获取锁之后执行
if (!isheldexclusively())
throw new illegalmonitorstateexception();
// 条件队列的头节点
node first = firstwaiter;
// 如果有等待条件的节点，则通知它条件已成立
if (first != null)
dosignal(first);
}
// abstractqueuedsynchronizer.conditionobject.dosignal
private void dosignal(node first) {
do {
// 移到条件队列的头节点往后一位
if ( (firstwaiter = first.nextwaiter) == null)
lastwaiter = null;
// 相当于把头节点从队列中出队
first.nextwaiter = null;
// 转移节点到aqs队列中
} while (!transferforsignal(first) &&
(first = firstwaiter) != null);
}
// abstractqueuedsynchronizer.transferforsignal
final boolean transferforsignal(node node) {
// 把节点的状态更改为0，也就是说即将移到aqs队列中
// 如果失败了，说明节点已经被改成取消状态了
// 返回false，通过上面的循环可知会寻找下一个可用节点
if (!compareandsetwaitstatus(node, node.condition, 0))
return false;
// 调用aqs的入队方法把节点移到aqs的队列中
// 注意，这里enq()的返回值是node的上一个节点，也就是旧尾节点
node p = enq(node);
// 上一个节点的等待状态
int ws = p.waitstatus;
// 如果上一个节点已取消了，或者更新状态为signal失败（也是说明上一个节点已经取消了）
// 则直接唤醒当前节点对应的线程
if (ws > 0 || !compareandsetwaitstatus(p, ws, node.signal))
locksupport.unpark(node.thread);
// 如果更新上一个节点的等待状态为signal成功了
// 则返回true，这时上面的循环不成立了，退出循环，也就是只通知了一个节点
// 此时当前节点还是阻塞状态
// 也就是说调用signal()的时候并不会真正唤醒一个节点
// 只是把节点从条件队列移到aqs队列中
return true;
}

signal()方法的大致流程为：

（1）从条件队列的头节点开始寻找一个非取消状态的节点；

（2）把它从条件队列移到aqs队列；

（3）且只移动一个节点；

注意，这里调用signal()方法后并不会真正唤醒一个节点，那么，唤醒一个节点是在啥时候呢？

还记得开头例子吗？倒回去再好好看看，signal()方法后，最终会执行lock.unlock()方法，此时才会真正唤醒一个节点，唤醒的这个节点如果曾经是条件节点的话又会继续执行await()方法“分界线”下面的代码。

结束了，仔细体会下^^

如果非要用一个图来表示的话，我想下面这个图可以大致表示一下（这里是用时序图画的，但是实际并不能算作一个真正的时序图哈，了解就好）：

总结

（1）重入锁是指可重复获取的锁，即一个线程获取锁之后再尝试获取锁时会自动获取锁；

（2）在reentrantlock中重入锁是通过不断累加state变量的值实现的；

（3）reentrantlock的释放要跟获取匹配，即获取了几次也要释放几次；

（4）reentrantlock默认是非公平模式，因为非公平模式效率更高；

（5）条件锁是指为了等待某个条件出现而使用的一种锁；

（6）条件锁比较经典的使用场景就是队列为空时阻塞在条件notempty上；

（7）reentrantlock中的条件锁是通过aqs的conditionobject内部类实现的；

（8）await()和signal()方法都必须在获取锁之后释放锁之前使用；

（9）await()方法会新建一个节点放到条件队列中，接着完全释放锁，然后阻塞当前线程并等待条件的出现；

（10）signal()方法会寻找条件队列中第一个可用节点移到aqs队列中；

（11）在调用signal()方法的线程调用unlock()方法才真正唤醒阻塞在条件上的节点（此时节点已经在aqs队列中）；

（12）之后该节点会再次尝试获取锁，后面的逻辑与lock()的逻辑基本一致了。

彩蛋

为什么java有自带的关键字synchronized了还需要实现一个reentrantlock呢？

首先，它们都是可重入锁；

其次，它们都默认是非公平模式；

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持。