问题
(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 呢?
首先, 它们都是可重入锁;
其次, 它们都默认是非公平模式;
然后,..., 呃, 我们下一章继续深入探讨 ReentrantLock VS synchronized.
来源: https://www.cnblogs.com/tong-yuan/p/ReentrantLock-Condition.html