共享模式acquire实现流程
上文我们讲解了AbstractQueuedSynchronizer独占模式的acquire实现流程,本文趁热打铁继续看一下AbstractQueuedSynchronizer共享模式acquire的实现流程。连续两篇文章的学习,也可以对比独占模式acquire和共享模式acquire的区别,加深对于AbstractQueuedSynchronizer的理解。
先看一下共享模式acquire的实现,方法为acquireShared和acquireSharedInterruptibly,两者差别不大,区别就在于后者有中断处理,以acquireShared为例:
1 | 1 public final void acquireShared(int arg) { 2 if (tryAcquireShared(arg) < 0) 3 doAcquireShared(arg); 4 }
|
Salin selepas log masuk
这里就能看出第一个差别来了:独占模式acquire的时候子类重写的方法tryAcquire返回的是boolean,即是否tryAcquire成功;共享模式acquire的时候,返回的是一个int型变量,判断是否<0。doAcquireShared方法的实现为:
1 | 1 private void doAcquireShared(int arg) { 2 final Node node = addWaiter(Node.SHARED); 3 boolean failed = true; 4 try { 5 boolean interrupted = false; 6 for (;;) { 7 final Node p = node.predecessor(); 8 if (p == head) { 9 int r = tryAcquireShared(arg);10 if (r >= 0) {11 setHeadAndPropagate(node, r);12 p.next = null;
|
Salin selepas log masuk
我们来分析一下这段代码做了什么:
-
addWaiter,把所有tryAcquireShared<0的线程实例化出一个Node,构建为一个FIFO队列,这和独占锁是一样的
拿当前节点的前驱节点,只有前驱节点是head的节点才能tryAcquireShared,这和独占锁也是一样的
前驱节点不是head的,执行"shouldParkAfterFailedAcquire() && parkAndCheckInterrupt()",for(;;)循环,"shouldParkAfterFailedAcquire()"方法执行2次,当前线程阻塞,这和独占锁也是一样的
确实,共享模式下的acquire和独占模式下的acquire大部分逻辑差不多,最大的差别在于tryAcquireShared成功之后,独占模式的acquire是直接将当前节点设置为head节点即可,共享模式会执行setHeadAndPropagate方法,顾名思义,即在设置head之后多执行了一步propagate操作。setHeadAndPropagate方法源码为:
1 | 1 private void setHeadAndPropagate(Node node, int propagate) { 2 Node h = head;
|
Salin selepas log masuk
第3行的代码设置重设head,第2行的代码由于第3行的代码要重设head,因此先定义一个Node型变量h获得原head的地址,这两行代码很简单。
第19行~第23行的代码是独占锁和共享锁最不一样的一个地方,我们再看独占锁acquireQueued的代码:
1 | 1 final boolean acquireQueued( final Node node, int arg) { 2 boolean failed = true; 3 try { 4 boolean interrupted = false; 5 for (;;) { 6 final Node p = node.predecessor(); 7 if (p == head && tryAcquire(arg)) { 8 setHead(node); 9 p.next = null;
|
Salin selepas log masuk
这意味着独占锁某个节点被唤醒之后,它只需要将这个节点设置成head就完事了,而共享锁不一样,某个节点被设置为head之后,如果它的后继节点是SHARED状态的,那么将继续通过doReleaseShared方法尝试往后唤醒节点,实现了共享状态的向后传播。
共享模式release实现流程
上面讲了共享模式下acquire是如何实现的,下面再看一下release的实现流程,方法为releaseShared:
1 | 1 public final boolean releaseShared(int arg) {2 if (tryReleaseShared(arg)) {3 doReleaseShared();4 return true;5 }6 return false;7 }
|
Salin selepas log masuk
tryReleaseShared方法是子类实现的,如果tryReleaseShared成功,那么执行doReleaseShared()方法:
1 | 1 private void doReleaseShared() { 2 13 for (;;) {14 Node h = head;15 if (h != null && h != tail) {16 int ws = h.waitStatus;17 if (ws == Node.SIGNAL) {18 if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))19 continue ;
|
Salin selepas log masuk
主要是两层逻辑:
头结点本身的waitStatus是SIGNAL且能通过CAS算法将头结点的waitStatus从SIGNAL设置为0,唤醒头结点的后继节点
头结点本身的waitStatus是0的话,尝试将其设置为PROPAGATE状态的,意味着共享状态可以向后传播
Condition的await()方法实现原理----构建等待队列
我们知道,Condition是用于实现通知/等待机制的,和Object的wait()/notify()一样,由于本文之前描述AbstractQueuedSynchronizer的共享模式的篇幅不是很长,加之Condition也是AbstractQueuedSynchronizer的一部分,因此将Condition也放在这里写了。
Condition分为await()和signal()两部分,前者用于等待、后者用于唤醒,首先看一下await()是如何实现的。Condition本身是一个接口,其在AbstractQueuedSynchronizer中的实现为ConditionObject:
1 | 1 public class ConditionObject implements Condition, java.io.Serializable {2 private static final long serialVersionUID = 1173984872572414699L;3 4 private transient Node firstWaiter;5 6 private transient Node lastWaiter;7 8 ...9 }
|
Salin selepas log masuk
这里贴了一些字段定义,后面都是方法就不贴了,会对重点方法进行分析的。从字段定义我们可以看到,ConditionObject全局性地记录了第一个等待的节点与最后一个等待的节点。
像ReentrantLock每次要使用ConditionObject,直接new一个ConditionObject出来即可。我们关注一下await()方法的实现:
1 | 1 public final void await() throws InterruptedException { 2 if (Thread.interrupted()) 3 throw new InterruptedException(); 4 Node node = addConditionWaiter(); 5 int savedState = fullyRelease(node); 6 int interruptMode = 0; 7 while (!isOnSyncQueue(node)) { 8 LockSupport.park(this); 9 if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)10 break ;11 }12 if (acquireQueued(node, savedState) && interruptMode != THROW_IE)13 interruptMode = REINTERRUPT;14 if (node.nextWaiter != null)
|
Salin selepas log masuk
Salin selepas log masuk
第2行~第3行的代码用于处理中断,第4行代码比较关键,添加Condition的等待者,看一下实现:
1 | 1 private Node addConditionWaiter() { 2 Node t = lastWaiter; 3
|
Salin selepas log masuk
首先拿到队列(注意数据结构,Condition构建出来的也是一个队列)中最后一个等待者,紧接着第4行的的判断,判断最后一个等待者的waitStatus不是CONDITION的话,执行第5行的代码,解绑取消的等待者,因为通过第8行的代码,我们看到,new出来的Node的状态都是CONDITION的。
那么unlinkCancelledWaiters做了什么?里面的流程就不看了,就是一些指针遍历并判断状态的操作,总结一下就是:从头到尾遍历每一个Node,遇到Node的waitStatus不是CONDITION的就从队列中踢掉,该节点的前后节点相连。
接着第8行的代码前面说过了,new出来了一个Node,存储了当前线程,waitStatus是CONDITION,接着第9行~第13行的操作很好理解:
如果lastWaiter是null,说明FIFO队列中没有任何Node,firstWaiter=Node
如果lastWaiter不是null,说明FIFO队列中有Node,原lastWaiter的next指向Node
无论如何,新加入的Node编程lastWaiter,即新加入的Node一定是在最后面
用一张图表示一下构建的数据结构就是:

对比学习,我们总结一下Condition构建出来的队列和AbstractQueuedSynchronizer构建出来的队列的差别,主要体现在2点上:
AbstractQueuedSynchronizer构建出来的队列,头节点是一个没有Thread的空节点,其标识作用,而Condition构建出来的队列,头节点就是真正等待的节点
AbstractQueuedSynchronizer构建出来的队列,节点之间有next与pred相互标识该节点的前一个节点与后一个节点的地址,而Condition构建出来的队列,只使用了nextWaiter标识下一个等待节点的地址
整个过程中,我们看到没有使用任何CAS操作,firstWaiter和lastWaiter也没有用volatile修饰,其实原因很简单:要await()必然要先lock(),既然lock()了就表示没有竞争,没有竞争自然也没必要使用volatile+CAS的机制去保证什么。
Condition的await()方法实现原理----线程等待
前面我们看了Condition构建等待队列的过程,接下来我们看一下等待的过程,await()方法的代码比较短,再贴一下:
1 | 1 public final void await() throws InterruptedException { 2 if (Thread.interrupted()) 3 throw new InterruptedException(); 4 Node node = addConditionWaiter(); 5 int savedState = fullyRelease(node); 6 int interruptMode = 0; 7 while (!isOnSyncQueue(node)) { 8 LockSupport.park(this); 9 if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)10 break ;11 }12 if (acquireQueued(node, savedState) && interruptMode != THROW_IE)13 interruptMode = REINTERRUPT;14 if (node.nextWaiter != null)
|
Salin selepas log masuk
Salin selepas log masuk
构建完毕队列之后,执行第5行的fullyRelease方法,顾名思义:fullyRelease方法的作用是完全释放Node的状态。方法实现为:
1 | 1 final int fullyRelease(Node node) { 2 boolean failed = true; 3 try { 4 int savedState = getState(); 5 if (release(savedState)) { 6 failed = false; 7 return savedState; 8 } else { 9 throw new IllegalMonitorStateException();10 }11 } finally {12 if (failed)13 node.waitStatus = Node.CANCELLED;14 }15 }
|
Salin selepas log masuk
这里第4行获取state,第5行release的时候将整个state传过去,理由是某线程可能多次调用了lock()方法,比如调用了10次lock,那么此线程就将state加到了10,所以这里要将10传过去,将状态全部释放,这样后面的线程才能重新从state=0开始竞争锁,这也是方法被命名为fullyRelease的原因,因为要完全释放锁,释放锁之后,如果有竞争锁的线程,那么就唤醒第一个,这都是release方法的逻辑了,前面的文章详细讲解过。
接着看await()方法的第7行判断"while(!isOnSyncQueue(node))":
1 | 1 final boolean isOnSyncQueue(Node node) { 2 if (node.waitStatus == Node.CONDITION || node.prev == null) 3 return false; 4 if (node.next != null)
|
Salin selepas log masuk
注意这里的判断是Node是否在AbstractQueuedSynchronizer构建的队列中而不是Node是否在Condition构建的队列中,如果Node不在AbstractQueuedSynchronizer构建的队列中,那么调用LockSupport的park方法阻塞。
至此调用await()方法的线程构建Condition等待队列--释放锁--等待的过程已经全部分析完毕。
Condition的signal()实现原理
上面的代码分析了构建Condition等待队列--释放锁--等待的过程,接着看一下signal()方法通知是如何实现的:
1 | 1 public final void signal() {2 if (!isHeldExclusively())3 throw new IllegalMonitorStateException();4 Node first = firstWaiter;5 if (first != null)6 doSignal(first);7 }
|
Salin selepas log masuk
首先从第2行的代码我们看到,要能signal(),当前线程必须持有独占锁,否则抛出异常IllegalMonitorStateException。
那么真正操作的时候,获取第一个waiter,如果有waiter,调用doSignal方法:
1 | 1 private void doSignal(Node first) {2 do {3 if ( (firstWaiter = first.nextWaiter) == null)4 lastWaiter = null;5 first.nextWaiter = null;6 } while (!transferForSignal(first) &&7 (first = firstWaiter) != null);8 }
|
Salin selepas log masuk
第3行~第5行的代码很好理解:
重新设置firstWaiter,指向第一个waiter的nextWaiter
如果第一个waiter的nextWaiter为null,说明当前队列中只有一个waiter,lastWaiter置空
因为firstWaiter是要被signal的,因此它没什么用了,nextWaiter置空
接着执行第6行和第7行的代码,这里重点就是第6行的transferForSignal方法:
1 2 3 4 5 6 7 8 9 10 11 12 13 | 1 final boolean transferForSignal(Node node) { 2 5 if (!compareAndSetWaitStatus(node, Node.CONDITION, 0)) 6 return false; 7 8 14 Node p = enq(node);15 int ws = p.waitStatus;16 if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))17 LockSupport.unpark(node.thread);18 return true;19 }</p>
<p><span style= "font-size: 13px; font-family: 宋体" >方法本意是将一个节点从Condition队列转换为AbstractQueuedSynchronizer队列,总结一下方法的实现:</span></p>
<ol class = " list-paddingleft-2" >
<li><p><span style= "font-size: 13px; font-family: 宋体" >尝试将Node的waitStatus从CONDITION置为0,这一步失败直接返回false</span></p></li>
<li><p><span style= "font-size: 13px; font-family: 宋体" >当前节点进入调用enq方法进入AbstractQueuedSynchronizer队列</span></p></li>
<li><p><span style= "font-size: 13px; font-family: 宋体" >当前节点通过CAS机制将waitStatus置为SIGNAL</span></p></li>
</ol>
<p><span style= "font-size: 13px; font-family: 宋体" >最后上面的步骤全部成功,返回true,返回true唤醒等待节点成功。从唤醒的代码我们可以得出一个重要结论:<span style= "color: #ff0000" ><strong>某个await()的节点被唤醒之后并不意味着它后面的代码会立即执行,它会被加入到AbstractQueuedSynchronizer队列的尾部,只有前面等待的节点获取锁全部完毕才能轮到它</strong></span>。</span></p>
<p><span style= "font-size: 13px; font-family: 宋体" >代码分析到这里,我想类似的signalAll方法也没有必要再分析了,显然signalAll方法的作用就是将所有Condition队列中等待的节点逐一队列中从移除,由CONDITION状态变为SIGNAL状态并加入AbstractQueuedSynchronizer队列的尾部。</span></p>
<p> </p>
<p><strong><span style= 'font-size: 18px; font-family: "Microsoft YaHei"' >代码示例</span></strong></p>
<p><span style= "font-size: 13px; font-family: 宋体" >可能大家看了我分析半天代码会有点迷糊,这里最后我贴一段我用于验证上面Condition结论的示例代码,首先建立一个Thread,我将之命名为ConditionThread:</span></p>
<div class = "cnblogs_code" ><pre class = "brush:php;toolbar:false" > 1 4 public class ConditionThread implements Runnable { 5 6 private Lock lock; 7 8 private Condition condition; 9 10 public ConditionThread(Lock lock, Condition condition) {11 this.lock = lock;12 this.condition = condition;13 }14 15 @Override16 public void run() {17 18 if ( "线程0" .equals(JdkUtil.getThreadName())) {19 thread0Process();20 } else if ( "线程1" .equals(JdkUtil.getThreadName())) {21 thread1Process();22 } else if ( "线程2" .equals(JdkUtil.getThreadName())) {23 thread2Process();24 }25 26 }27 28 private void thread0Process() {29 try {30 lock.lock();31 System.out.println( "线程0休息5秒" );32 JdkUtil.sleep(5000);33 condition.signal();34 System.out.println( "线程0唤醒等待线程" );35 } finally {36 lock.unlock();37 }38 }39 40 private void thread1Process() {41 try {42 lock.lock();43 System.out.println( "线程1阻塞" );44 condition.await();45 System.out.println( "线程1被唤醒" );46 } catch (InterruptedException e) {47 48 } finally {49 lock.unlock();50 }51 }52 53 private void thread2Process() {54 try {55 System.out.println( "线程2想要获取锁" );56 lock.lock();57 System.out.println( "线程2获取锁成功" );58 } finally {59 lock.unlock();60 }61 }62 63 }
|
Salin selepas log masuk
这个类里面的方法就不解释了,反正就三个方法片段,根据线程名判断,每个线层执行的是其中的一个代码片段。写一段测试代码:
1 | 1 4 @Test 5 public void testCondition() throws Exception { 6 Lock lock = new ReentrantLock(); 7 Condition condition = lock.newCondition(); 8 9
|
Salin selepas log masuk
测试代码的意思是:
线程1先启动,获取锁,调用await()方法等待
线程0后启动,获取锁,休眠5秒准备signal()
线程2最后启动,获取锁,由于线程0未使用完毕锁,因此线程2排队,可以此时由于线程0还未signal(),因此线程1在线程0执行signal()后,在AbstractQueuedSynchronizer队列中的顺序是在线程2后面的
代码执行结果为:
1 | <span style= "color: #008080" > 1</span> <span style= "color: #000000" >线程1阻塞</span><span style= "color: #008080" > 2</span> <span style= "color: #000000" >线程0休息5秒</span><span style= "color: #008080" > 3</span> <span style= "color: #000000" >线程2想要获取锁</span><span style= "color: #008080" > 4</span> <span style= "color: #000000" >线程0唤醒等待线程</span><span style= "color: #008080" > 5</span> <span style= "color: #000000" >线程2获取锁成功</span><span style= "color: #008080" > 6</span> <span style= "color: #000000" >线程1被唤醒</span><span style= "color: #008080" ><br></span>
|
Salin selepas log masuk
符合我们的结论:signal()并不意味着被唤醒的线程立即执行。由于线程2先于线程0排队,因此看到第5行打印的内容,线程2先获取锁。
Atas ialah kandungan terperinci 共享模式与基于Condition的等待/通知实例详解. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!