本篇博客中的大部分内容并非原创,仅做整理作为学习资料使用
简介
条件锁,是指在获取锁之后发现当前业务场景自己无法处理,而需要等待某个条件的出现才可以继续处理时使用的一种锁。
比如,在阻塞队列中,当队列中没有元素的时候是无法弹出一个元素的,这时候就需要阻塞在条件notEmpty上,等待其它线程往里面放入一个元素后,唤醒这个条件notEmpty,当前线程才可以继续去做“弹出一个元素”的行为。
注意,这里的条件,必须是在获取锁之后去等待,对应到ReentrantLock的条件锁,就是获取锁之后才能调用condition.await()方法。
在java中,条件锁的实现都在AQS的ConditionObject类中,ConditionObject实现了Condition接口,下面我们通过一个例子来进入到条件锁的学习中。
使用示例
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38
| public class ReentrantLockTest { public static void main(String[] args) throws InterruptedException { ReentrantLock lock = new ReentrantLock(); Condition condition = lock.newCondition();
new Thread(()->{ try { lock.lock(); try { System.out.println("before await"); condition.await(); System.out.println("after await"); } finally { lock.unlock(); } } catch (InterruptedException e) { e.printStackTrace(); } }).start(); Thread.sleep(1000); lock.lock(); try { Thread.sleep(2000); System.out.println("before signal"); condition.signal(); System.out.println("after signal"); } finally { lock.unlock(); } } }
|
上面的代码很简单,一个线程等待条件,另一个线程通知条件已成立,后面的数字代表代码实际运行的顺序,如果你能把这个顺序看懂基本条件锁掌握得差不多了。
源码分析
ConditionObject的主要属性
1 2 3 4 5 6
| public class ConditionObject implements Condition, java.io.Serializable { private transient Node firstWaiter; private transient Node lastWaiter; }
|
可以看到条件锁中也维护了一个队列,为了和AQS的队列区分,我这里称为条件队列,firstWaiter是队列的头节点,lastWaiter是队列的尾节点,它们是干什么的呢?接着看。
lock.newCondition()方法
新建一个条件锁。
1 2 3 4 5 6 7 8 9 10
| public Condition newCondition() { return sync.newCondition(); }
final ConditionObject newCondition() { return new ConditionObject(); }
public ConditionObject() { }
|
新建一个条件锁最后就是调用的AQS中的ConditionObject类来实例化条件锁。
condition.await()方法
condition.await()方法,表明现在要等待条件的出现。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89
| public final void await() throws InterruptedException { if (Thread.interrupted()) throw new InterruptedException(); Node node = addConditionWaiter(); int savedState = fullyRelease(node); int interruptMode = 0; while (!isOnSyncQueue(node)) { LockSupport.park(this); if ((interruptMode = checkInterruptWhileWaiting(node)) != 0) break; } if (acquireQueued(node, savedState) && interruptMode != THROW_IE) interruptMode = REINTERRUPT; if (node.nextWaiter != null) unlinkCancelledWaiters(); if (interruptMode != 0) reportInterruptAfterWait(interruptMode); }
private Node addConditionWaiter() { Node t = lastWaiter; if (t != null && t.waitStatus != Node.CONDITION) { unlinkCancelledWaiters(); t = lastWaiter; } Node node = new Node(Thread.currentThread(), Node.CONDITION); if (t == null) firstWaiter = node; else t.nextWaiter = node; lastWaiter = node; return node; }
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; } }
final boolean isOnSyncQueue(Node node) { if (node.waitStatus == Node.CONDITION || node.prev == null) return false; if (node.next != null) return true; return findNodeFromTail(node); }
|
这里有几个难理解的点:
(1)Condition的队列和AQS的队列不完全一样;
1 2 3
| AQS的队列头节点是不存在任何值的,是一个虚节点;
Condition的队列头节点是存储着实实在在的元素值的,是真实节点。
|
(2)各种等待状态(waitStatus)的变化;
1 2 3 4 5 6 7 8 9
| 首先,在条件队列中,新建节点的初始等待状态是CONDITION(-2);
其次,移到AQS的队列中时等待状态会更改为0(AQS队列节点的初始等待状态为0);
然后,在AQS的队列中如果需要阻塞,会把它上一个节点的等待状态设置为SIGNAL(-1);
最后,不管在Condition队列还是AQS队列中,已取消的节点的等待状态都会设置为CANCELLED(1);
另外,后面我们在共享锁的时候还会讲到另外一种等待状态叫PROPAGATE(-3)。
|
(3)相似的名称;
1 2 3
| AQS中下一个节点是next,上一个节点是prev;
Condition中下一个节点是nextWaiter,没有上一个节点。
|
如果弄明白了这几个点,看懂上面的代码还是轻松加愉快的,如果没弄明白,彤哥这里指出来了,希望您回头再看看上面的代码。
下面总结一下await()方法的大致流程:
(1)新建一个节点加入到条件队列中去;
(2)完全释放当前线程占有的锁;
(3)阻塞当前线程,并等待条件的出现;
(4)条件已出现(此时节点已经移到AQS的队列中),尝试获取锁;
也就是说await()方法内部其实是先释放锁->等待条件->再次获取锁
的过程。
condition.signal()方法
condition.signal()方法通知条件已经出现。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48
| public final void signal() { if (!isHeldExclusively()) throw new IllegalMonitorStateException(); Node first = firstWaiter; if (first != null) doSignal(first); }
private void doSignal(Node first) { do { if ( (firstWaiter = first.nextWaiter) == null) lastWaiter = null; first.nextWaiter = null; } while (!transferForSignal(first) && (first = firstWaiter) != null); }
final boolean transferForSignal(Node node) { if (!compareAndSetWaitStatus(node, Node.CONDITION, 0)) return false;
Node p = enq(node); int ws = p.waitStatus; if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL)) LockSupport.unpark(node.thread); 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()的逻辑基本一致了。
参考