装修网站模板下载,网站做后台,wordpress 小游戏,尚层装饰官网一、概述
条件锁就是指在获取锁之后发现当前业务场景自己无法处理#xff0c;而需要等待某个条件的出现才可以继续处理时使用的一种锁。
比如#xff0c;在阻塞队列中#xff0c;当队列中没有元素的时候是无法弹出一个元素的#xff0c;这时候就需要阻塞在条件notEmpty上…一、概述
条件锁就是指在获取锁之后发现当前业务场景自己无法处理而需要等待某个条件的出现才可以继续处理时使用的一种锁。
比如在阻塞队列中当队列中没有元素的时候是无法弹出一个元素的这时候就需要阻塞在条件notEmpty上等待其它线程往里面放入一个元素后唤醒这个条件notEmpty当前线程才可以继续去做“弹出一个元素”的行为。
注意这里的条件必须是在获取锁之后去等待对应到ReentrantLock的条件锁就是获取锁之后才能调用condition.await()方法。
在Java中条件锁的实现都在AQS的ConditionObject类中ConditionObject实现了Condition接口所以ReentrantLock的条件锁是基于AQS实现的。 sychronized Object.wait Lock 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(); // 1try {System.out.println(before await); // 2// 等待条件 // 该线程执行到这里就会进入到阻塞状态直到达到了条件后// 由其他线程执行signal()方法来告知该线程已经达到条件了// 该线程就会在这里被唤醒继续向下执行condition.await(); // 3System.out.println(after await); // 10} finally {// 释放锁lock.unlock(); // 11}} catch (InterruptedException e) {e.printStackTrace();}}).start();// 这里睡1000ms是为了让上面的线程先获取到锁Thread.sleep(1000);// main方法的当前线程来获取锁该线程就是用来进行相关业务处理进而达到条件锁条件的。// 可以看作是一个生产者lock.lock(); // 4try {// 这里睡2000ms代表这个线程执行业务需要的时间// 当完成这里的2000ms之后就认为是符合条件锁条件了Thread.sleep(2000); // 5System.out.println(before signal); // 6// 通知条件已成立condition.signal(); // 7System.out.println(after signal); // 8} finally {// 释放锁lock.unlock(); // 9}}
}上面的代码很简单一个线程等待条件另一个线程通知条件已成立后面的数字代表代码实际运行的顺序。
由上面的例子我们也可以看出await()和signal()方法都必须在获取锁之后释放锁之前使用
三、源码分析
3.1 Condition接口
Condition是一个接口从1.5的时候出现的是用来替代Object的wait、notify。所以显而易见Condition的await和signal肯定效率更高、安全性更好。Condition是依赖于lock实现的。并且await和signal只能在lock的保护之内使用。
package java.util.concurrent.locks;import java.util.concurrent.TimeUnit;
import java.util.Date;public interface Condition {//导致当前线程等到发信号或 interrupted 。void await() throws InterruptedException;//使当前线程等待直到发出信号void awaitUninterruptibly();//使当前线程等待直到发出信号或中断或指定的等待时间过去。 long awaitNanos(long nanosTimeout) throws InterruptedException;//使当前线程等待直到发出信号或中断或指定的等待时间过去。 boolean await(long time, TimeUnit unit) throws InterruptedException;//使当前线程等待直到发出信号或中断或者指定的最后期限过去。 boolean awaitUntil(Date deadline) throws InterruptedException;//唤醒一个等待线程。 void signal();//唤醒所有等待线程。 void signalAll();
}3.2 内部类
ConditionObject是Condition的实现类。
//AbstractQueuedSynchronizer.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是队列的尾节点。
ConditionObject中的条件队列和AQS中的同步队列使用的是相同的节点类型Node但是两个队列还是有一些不同的在后续会详细讲解。
3.3 lock.newCondition()方法
新建一个条件锁。
ReentrantLock.newCondition()ReentrantLock.Sync.newCondition()AbstractQueuedSynchronizer.ConditionObject.ConditionObject()
public class ReentrantLock implements Lock, java.io.Serializable {private final Sync sync;// 创建条件锁public Condition newCondition() {return sync.newCondition();}/*** 抽象内部类*/abstract static class Sync extends AbstractQueuedSynchronizer {// 条件锁final ConditionObject newCondition() {return new ConditionObject();}}
}// AbstractQueuedSynchronizer.ConditionObject.ConditionObject()
public ConditionObject() { }新建一个条件锁最后就是调用的AQS中的ConditionObject类来实例化条件锁。
3.4 condition.await()方法
condition.await()方法表明现在要等待条件的出现只有满足条件之后获取锁的线程才可以继续向后执行。
await()方法会新建一个节点放到条件队列中接着完全释放锁然后阻塞当前线程并等待条件的出现
// AbstractQueuedSynchronizer.ConditionObject.await()
public final void await() throws InterruptedException {// 如果线程中断了抛出异常if (Thread.interrupted())throw new InterruptedException();// 添加节点到Condition的队列中并返回该节点Node node addConditionWaiter();// 完全释放当前线程获取的锁// 因为锁是可重入的所以这里要把获取的锁全部释放int savedState fullyRelease(node);/*** 中断标志用来标识线程是否是被中断唤醒的* interruptMode 0表示不是被中断唤醒的* interruptMode ! 0表示是被中断唤醒的* interruptMode REINTERRUPT 1表示当前线程在其他线程调用signal()之后被中断唤醒* interruptMode THROW_IE -1表示当前线程在其他线程调用signal()之前被中断唤醒*/int interruptMode 0;// 是否在同步队列中如果该线程节点从条件队列移出到同步队列中// 说明当前已经满足条件线程已经被唤醒则跳出循环while (!isOnSyncQueue(node)) {// 阻塞当前线程LockSupport.park(this);// 上面部分是调用await()时释放自己占有的锁并阻塞自己等待条件的出现// *************************分界线************************* //// 下面部分是条件已经出现该线程被唤醒尝试重新去获取锁继续向后执行// checkInterruptWhileWaiting()中会判断当前线程是否是被中断唤醒的// 返回值非0表示是被中断唤醒的会通过break跳出while。// 因为如果是中断唤醒的有可能实际上该线程还没有等到条件满足的时候就被唤醒了// 这样该线程的Node节点一定没有被转移到同步队列中所以就不可能通过循环条件来跳出循环。// 只能是我们手动调用break来跳出循环毕竟await()方法还是要响应中断的// 不能在其他线程已经发出中断信号后还让线程在这个循环里自选而不响应中断。if ((interruptMode checkInterruptWhileWaiting(node)) ! 0)break;}// 尝试获取锁注意第二个参数这里需要获取所的数量要和该线程最初持有锁的数量相同// 使该线程恢复到最开始的持有锁的状态/*** 我们从两个条件分别去分析* 1、acquireQueued(node, savedState)* 尝试获取锁注意第二个参数这里需要获取所的数量要和该线程最初持有锁的数量相同* 使该线程恢复到最开始的持有锁的状态* 该方法在之前ReentrantLock章节中已经讲解过了在这个方法中线程就会不断地尝试获取锁* 如果没获取到就会再次阻塞在这个位置等到被唤醒后继续抢占锁直到成功获取锁之后该方法才会返回。* 该方法的返回值是中断标记如果该线程是被中断信号唤醒并且抢占到锁的就会返回true* 如果线程不是被中断唤醒的则返回false。* * 2、interruptMode ! THROW_IE* interruptMode标识的是线程在等待条件满足被阻塞的过程当中是否是被中断唤醒的* 这里的条件interruptMode ! THROW_IE表示该线程不是在其他线程调用signal()之前被中断唤醒的* 所以第一个条件是表示该线程在重新获取锁的时被阻塞然后又被唤醒重新获取到锁的过程中是不是被中断唤醒的* 第二个条件表示该线程被阻塞等待条件满足的过程当中是不是被中断唤醒的* 第一个条件为true说明该线程是在重新获取锁的过程中接收到过中断信号也就说是这一次中断是在该线程等待* 条件被阻塞然后又被唤醒之后才发生的这个中断信号一定是发生在其他线程调用signal()之后* 第二个条件为true说明该线程在其他线程调用signal之前没有接收到过中断信号* 当两个条件都为true时就会进入到if代码块中将interruptMode REINTERRUPT也就是该线程收到了中断信号* 并且是在其他线程调用signal()之后收到的中断信号* 如果该线程实在signal之前被中断的那么该线程的流程就到此结束了需要从头再来获取锁* 就不能执行后续的步骤了也就不能进入到该if代码块中*/// 如果没获取到会再次阻塞if (acquireQueued(node, savedState) interruptMode ! THROW_IE)interruptMode REINTERRUPT;// 清除取消的节点if (node.nextWaiter ! null) // clean up if cancelledunlinkCancelledWaiters();// 线程中断相关if (interruptMode ! 0)reportInterruptAfterWait(interruptMode);
}// AbstractQueuedSynchronizer.ConditionObject.addConditionWaiter
// 为线程创建Node节点并将其添加到条件队列当中去
private Node addConditionWaiter() {Node t lastWaiter;/*** 如果条件队列的尾节点已取消非Node.CONDITION状态从头节点开始清除所有已取消的节点* * 该条件成立的例子* 1、假设有两个线程thread0、thread1初始时thread0在未持有锁的情况下调用AQS.CO.await()* thread0执行到AQS.fullyRelease中时会将其对应节点的waitStatus字段设置为取消状态* 之后持有锁的thread1调用 AQS.CO.await()会执行到这里这是条件队列尾节点t.waitStatus为1* * 2、假设有两个线程thread0、thread1初始时thread0持有锁* 之后调用AQS.CO.await()释放锁并阻塞在LockSupport.park(this)处* 之后外部线程中断thread0thread0被唤醒后会执行到AQS.transferAfterCancelledWait的if处 * 将t.waitStatus设置为0之后thread1获取到锁执行到这里时t.waitStatus为0*/ if (t ! null t.waitStatus ! Node.CONDITION) {// 移除条件队列中所有不是Node.CONDITION状态的节点 unlinkCancelledWaiters();// 重新获取最新的尾节点经过unlinkCancelledWaiters()lastWaiter可能已经改变t lastWaiter;}// 新建一个节点它的等待状态是CONDITIONNode node new Node(Thread.currentThread(), Node.CONDITION);// 如果尾节点为空则把新节点赋值给头节点相当于初始化队列// 否则把新节点赋值给尾节点的nextWaiter指针if (t null)firstWaiter node;elset.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 queuereturn true;// 从AQS的尾节点开始往前寻找看是否可以找到当前节点找到了也说明已经在AQS的队列中了return findNodeFromTail(node);
}// AbstractQueuedSynchronizer.findNodeFromTail()
private boolean findNodeFromTail(Node node) {Node t tail;for (;;) {if (t node)return true;if (t null)return false;t t.prev;}
}// AbstractQueuedSynchronizer.ConditionObject.checkInterruptWhileWaiting()
// 根据节点的中断情况来返回其中断标志
private int checkInterruptWhileWaiting(Node node) {/*** 通过Thread.interrupted()来判断该线程是不是被中断如果没有被中断则返回0* 如果被中断了则在通过transferAfterCancelledWait(node)来判断该线程是在其他线程调用signal()前被中断* 还是调用signal()后被中断* 返回THROW_IE表示当前线程在其他线程调用signal前被中断* 返回REINTERRUPT表示当前线程在其他线程调用signal后被中断* 具体的分界点就是node.waitStatus的值若其值为Node.CONDITION是signal前被中断否则在signal后被中断*/return Thread.interrupted() ? (transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) : 0;
}// AbstractQueuedSynchronizer.ConditionObject.transferAfterCancelledWait()
// 返回true表示该节点是在调用signal()之前被中断的返回false表示是调用signal()之后被中断的
final boolean transferAfterCancelledWait(Node node) {/*** 如果node.waitStatus的值是Node.CONDITION说明该节点还没有被signal()方法唤醒* 也就是说该节点是在调用signal()前被中断的* * 例子* 该条件为true的情形* 假设仅有thread0thread0持有锁后调用AQS.CO.await()被park* 之后被外部线程中断会执行到这里此时条件为true* 该条件为false的情形* 假设有两个线程thread0、thread1thread0thread0持有锁后调用AQS.CO.await()被park* thread1获取到锁后调用AQS.CO.signal之后会执行到AQS.transferForSignal的第一个if处* 该if语句执行完后这里node.waitStatus被修改为0所以thread0执行这里的if语句时会失败*/if (compareAndSetWaitStatus(node, Node.CONDITION, 0)) {// 被中断后就需要重新进入同步队列抢占锁从头再来enq(node);return true;}// 到这里说明其他线程调用了signal()将当前线程对应的节点加入同步队列这里自旋等待入队完成while (!isOnSyncQueue(node))// 主动让出当前线程的CPU时间片Thread.yield(); // 到这里就说明是在调用signal()之后被中断的执行到这里的时候这个线程节点已经进入到了同步队列中了// 在后续的操作中就会执行acquireQueued(node, savedState)来不断尝试获取锁直到成功return false;
}// AbstractQueuedSynchronizer.ConditionObject.unlinkCancelledWaiters()
// 清除掉条件队列中所有被取消的节点
private void unlinkCancelledWaiters() {// 获取条件队列的头节点Node t firstWaiter;Node trail null;/*** 1、从firstWaiter开始将整个链表中t.waitStatus ! Node.CONDITION的节点移除掉* 2、节点移除后将其前置节点的nextWaiter指向后置节点。*/while (t ! null) {// 获取当前遍历到节点的下一个节点Node next t.nextWaiter;if (t.waitStatus ! Node.CONDITION) {/** 当前节点状态不是CONDITION* 将当前节点的nextWaiter设置为null。* 如果trail是空则将firstWaiter指向之前保存的t.nextWaiter,* 否则将trail.nextWaiter指向之前保存的t.nextWaiter。*/t.nextWaiter null;if (trail null)firstWaiter next;elsetrail.nextWaiter next;if (next null)lastWaiter trail;}elsetrail t;t next;}
}// AbstractQueuedSynchronizer.ConditionObject.reportInterruptAfterWait()
// 根据中断标识来进行不同的处理
private void reportInterruptAfterWait(int interruptMode) throws InterruptedException {// 如果当前线程在其他线程调用signal()之前被中断唤醒则直接抛出异常响应中断if (interruptMode THROW_IE)throw new InterruptedException();// 如果当前线程在其他线程调用signal()之后被中断唤醒这种情况不会对条件锁流程造成影响// 则设置中断标志但不会抛出异常中止执行else if (interruptMode REINTERRUPT)selfInterrupt();
}这里有几个难理解的点 Condition的队列和AQS的队列不完全一样 AQS的队列头节点是不存在任何值的是一个虚节点Condition的队列头节点是存储着实实在在的元素值的是真实节点。 各种等待状态(waitStatus)的变化 首先在条件队列中新建节点的初始等待状态是CONDITION(-2)其次移到AQS的队列中时等待状态会更改为0(AQS队列节点的初始等待状态为0)然后在AQS的队列中如果需要阻塞会把它上一个节点的等待状态设置为SIGNAL(-1)最后不管在Condition队列还是AQS队列中已取消的节点的等待状态都会设置为CANCELLED(1)另外后面我们在共享锁的时候还会讲到另外一种等待状态叫PROPAGATE(-3)。 相似的名称 AQS中下一个节点是next上一个节点是prevCondition中下一个节点是nextWaiter没有上一个节点。
总结一下await()方法的大致流程
新建一个节点加入到条件队列中去完全释放当前线程占有的锁阻塞当前线程并等待条件的出现条件已出现(此时节点已经移到AQS的队列中)尝试获取锁
也就是说await()方法内部其实是先释放锁-等待条件-再次获取锁的过程。
3.5 condition.signal()方法
condition.signal()方法通知条件已经出现。
这个方法是由获取锁的其他线程执行的用来唤醒正在阻塞等待满足条件的线程。
signal()方法会寻找条件队列中第一个可用节点移到AQS队列中
// 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)// 如果移动后firstWaiter为null说明已到队列尾部将lastWaiter设置为nulllastWaiter null;// 将first与其后继节点断开相当于把头节点从队列中出队first.nextWaiter null;// 转移节点到AQS队列中// 条件1只要一个线程转移到AQS同步队列成功transferForSignal返回true就会终止该循环// 条件2用于判断是否遍历到了队列尾部} 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()方法的大致流程为
从条件队列的头节点开始寻找一个非取消状态的节点把它从条件队列移到AQS队列且只移动一个节点
注意这里调用signal()方法后并不会真正唤醒一个节点那么唤醒一个节点是在啥时候呢
我们可以再回去看一下本文最开始的使用案例在其他线程调用signal()方法后该线程最终会执行lock.unlock()方法此时才会真正唤醒一个正在同步队列中的节点唤醒的这个节点如果曾经是条件节点被转移到条件队列中的话就会继续执行await()方法“分界线”下面的代码。也就是说在调用signal()方法的线程调用unlock()方法才是真正唤醒阻塞在条件上的节点此时节点已经在AQS队列中被唤醒之后被唤醒的节点会再次尝试获取锁后面的逻辑与lock()的逻辑基本一致了。
3.6 condition.signalAll()方法
signalAll与signal类似区别是signalAll会将所有节点加入同步队列除了doSignalAll方法以外其他的方法都是一样的
private void doSignalAll(Node first) {// 将条件队列的头节点和尾节点都置为nulllastWaiter firstWaiter null;// 遍历条件队列依次唤醒所有线程do {Node next first.nextWaiter;first.nextWaiter null;transferForSignal(first);first next;} while (first ! null);
}四、总结
条件锁是指为了等待某个条件出现而使用的一种锁条件锁比较经典的使用场景就是队列为空时阻塞在条件notEmpty上ReentrantLock中的条件锁是通过AQS的ConditionObject内部类实现的await()和signal()方法都必须在获取锁之后释放锁之前使用await()方法会新建一个节点放到条件队列中接着完全释放锁然后阻塞当前线程并等待条件的出现signal()方法会寻找条件队列中第一个可用节点移到AQS队列中在调用signal()方法的线程调用unlock()方法才真正唤醒阻塞在条件上的节点此时节点已经在AQS队列中之后该节点会再次尝试获取锁后面的逻辑与lock()的逻辑基本一致了。
参考文章
并发编程】Condition条件锁源码详解
