Java同步系列之ReentrantLock源码解析下(八)

devtool · · 146 次点击 · 开始浏览    置顶
## 问题 (1)条件锁是什么? (2)条件锁适用于什么场景? (3)条件锁的await()是在其它线程signal()的时候唤醒的吗? ## 简介 条件锁,是指在获取锁之后发现当前业务场景自己无法处理,而需要等待某个条件的出现才可以继续处理时使用的一种锁。 比如,在阻塞队列中,当队列中没有元素的时候是无法弹出一个元素的,这时候就需要阻塞在条件notEmpty上,等待其它线程往里面放入一个元素后,唤醒这个条件notEmpty,当前线程才可以继续去做“弹出一个元素”的行为。 注意,这里的条件,必须是**在获取锁之后去等待**,对应到ReentrantLock的条件锁,就是获取锁之后才能调用condition.await()方法。 在java中,条件锁的实现都在AQS的ConditionObject类中,ConditionObject实现了Condition接口,下面我们通过一个例子来进入到条件锁的学习中。 ## 使用示例 ```java 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的主要属性 ```java 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()方法 新建一个条件锁。 ```java // 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()方法,表明现在要等待条件的出现。 ```java // 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()方法通知条件已经出现。 ```java // 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()方法“分界线”下面的代码。 结束了,仔细体会下^^ 如果非要用一个图来表示的话,我想下面这个图可以大致表示一下(这里是用时序图画的,但是实际并不能算作一个真正的时序图哈,了解就好): ![ReentrantLock](https://gitee.com/alan-tang-tt/yuan/raw/master/死磕%20java同步系列/resource/condition.png) ## 总结 (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。
146 次点击  
加入收藏 微博
暂无回复
添加一条新回复 (您需要 登录 后才能回复 没有账号 ?)
  • 请尽量让自己的回复能够对别人有帮助
  • 支持 Markdown 格式, **粗体**、~~删除线~~、`单行代码`
  • 支持 @ 本站用户;支持表情(输入 : 提示),见 Emoji cheat sheet
  • 图片支持拖拽、截图粘贴等方式上传