Rumah > Java > javaTutorial > 共享模式与基于Condition的等待/通知实例详解

共享模式与基于Condition的等待/通知实例详解

零下一度
Lepaskan: 2017-07-17 14:17:25
asal
1698 orang telah melayarinya

共享模式acquire实现流程

上文我们讲解了AbstractQueuedSynchronizer独占模式的acquire实现流程,本文趁热打铁继续看一下AbstractQueuedSynchronizer共享模式acquire的实现流程。连续两篇文章的学习,也可以对比独占模式acquire和共享模式acquire的区别,加深对于AbstractQueuedSynchronizer的理解。

先看一下共享模式acquire的实现,方法为acquireShared和acquireSharedInterruptibly,两者差别不大,区别就在于后者有中断处理,以acquireShared为例:

 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 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; // help GC13                     if (interrupted)14                         selfInterrupt();15                     failed = false;16                     return;17                 }18             }19             if (shouldParkAfterFailedAcquire(p, node) &&20                 parkAndCheckInterrupt())21                 interrupted = true;22         }23     } finally {24         if (failed)25             cancelAcquire(node);26     }27 }
Salin selepas log masuk

我们来分析一下这段代码做了什么:

  1. addWaiter,把所有tryAcquireShared<0的线程实例化出一个Node,构建为一个FIFO队列,这和独占锁是一样的

  2. 拿当前节点的前驱节点,只有前驱节点是head的节点才能tryAcquireShared,这和独占锁也是一样的

  3. 前驱节点不是head的,执行"shouldParkAfterFailedAcquire() && parkAndCheckInterrupt()",for(;;)循环,"shouldParkAfterFailedAcquire()"方法执行2次,当前线程阻塞,这和独占锁也是一样的

确实,共享模式下的acquire和独占模式下的acquire大部分逻辑差不多,最大的差别在于tryAcquireShared成功之后,独占模式的acquire是直接将当前节点设置为head节点即可,共享模式会执行setHeadAndPropagate方法,顾名思义,即在设置head之后多执行了一步propagate操作。setHeadAndPropagate方法源码为:

 1 private void setHeadAndPropagate(Node node, int propagate) { 2     Node h = head; // Record old head for check below 3     setHead(node); 4     /* 5      * Try to signal next queued node if: 6      *   Propagation was indicated by caller, 7      *     or was recorded (as h.waitStatus) by a previous operation 8      *     (note: this uses sign-check of waitStatus because 9      *      PROPAGATE status may transition to SIGNAL.)10      * and11      *   The next node is waiting in shared mode,12      *     or we don&#39;t know, because it appears null13      *14      * The conservatism in both of these checks may cause15      * unnecessary wake-ups, but only when there are multiple16      * racing acquires/releases, so most need signals now or soon17      * anyway.18      */19     if (propagate > 0 || h == null || h.waitStatus < 0) {20         Node s = node.next;21         if (s == null || s.isShared())22             doReleaseShared();23     }24 }
Salin selepas log masuk

第3行的代码设置重设head,第2行的代码由于第3行的代码要重设head,因此先定义一个Node型变量h获得原head的地址,这两行代码很简单。

第19行~第23行的代码是独占锁和共享锁最不一样的一个地方,我们再看独占锁acquireQueued的代码:

 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; // help GC10                 failed = false;11                 return interrupted;12             }13             if (shouldParkAfterFailedAcquire(p, node) &&14                 parkAndCheckInterrupt())15                 interrupted = true;16         }17     } finally {18         if (failed)19             cancelAcquire(node);20     }21 }
Salin selepas log masuk

这意味着独占锁某个节点被唤醒之后,它只需要将这个节点设置成head就完事了,而共享锁不一样,某个节点被设置为head之后,如果它的后继节点是SHARED状态的,那么将继续通过doReleaseShared方法尝试往后唤醒节点,实现了共享状态的向后传播

共享模式release实现流程

上面讲了共享模式下acquire是如何实现的,下面再看一下release的实现流程,方法为releaseShared:

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 private void doReleaseShared() { 2     /* 3      * Ensure that a release propagates, even if there are other 4      * in-progress acquires/releases.  This proceeds in the usual 5      * way of trying to unparkSuccessor of head if it needs 6      * signal. But if it does not, status is set to PROPAGATE to 7      * ensure that upon release, propagation continues. 8      * Additionally, we must loop in case a new node is added 9      * while we are doing this. Also, unlike other uses of10      * unparkSuccessor, we need to know if CAS to reset status11      * fails, if so rechecking.12      */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;            // loop to recheck cases20                 unparkSuccessor(h);21             }22             else if (ws == 0 &&23                      !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))24                 continue;                // loop on failed CAS25         }26         if (h == head)                   // loop if head changed27             break;28     }29 }
Salin selepas log masuk

主要是两层逻辑:

  1. 头结点本身的waitStatus是SIGNAL且能通过CAS算法将头结点的waitStatus从SIGNAL设置为0,唤醒头结点的后继节点

  2. 头结点本身的waitStatus是0的话,尝试将其设置为PROPAGATE状态的,意味着共享状态可以向后传播

Condition的await()方法实现原理----构建等待队列

我们知道,Condition是用于实现通知/等待机制的,和Object的wait()/notify()一样,由于本文之前描述AbstractQueuedSynchronizer的共享模式的篇幅不是很长,加之Condition也是AbstractQueuedSynchronizer的一部分,因此将Condition也放在这里写了。

Condition分为await()和signal()两部分,前者用于等待、后者用于唤醒,首先看一下await()是如何实现的。Condition本身是一个接口,其在AbstractQueuedSynchronizer中的实现为ConditionObject:

1 public class ConditionObject implements Condition, java.io.Serializable {2         private static final long serialVersionUID = 1173984872572414699L;3         /** First node of condition queue. */4         private transient Node firstWaiter;5         /** Last node of condition queue. */6         private transient Node lastWaiter;7         8         ...9 }
Salin selepas log masuk

这里贴了一些字段定义,后面都是方法就不贴了,会对重点方法进行分析的。从字段定义我们可以看到,ConditionObject全局性地记录了第一个等待的节点与最后一个等待的节点

像ReentrantLock每次要使用ConditionObject,直接new一个ConditionObject出来即可。我们关注一下await()方法的实现:

 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) // clean up if cancelled15         unlinkCancelledWaiters();16     if (interruptMode != 0)17         reportInterruptAfterWait(interruptMode);18 }
Salin selepas log masuk
Salin selepas log masuk

第2行~第3行的代码用于处理中断,第4行代码比较关键,添加Condition的等待者,看一下实现:

 1 private Node addConditionWaiter() { 2     Node t = lastWaiter; 3     // If lastWaiter is cancelled, clean out. 4     if (t != null && t.waitStatus != Node.CONDITION) { 5         unlinkCancelledWaiters(); 6         t = lastWaiter; 7     } 8     Node node = new Node(Thread.currentThread(), Node.CONDITION); 9     if (t == null)10         firstWaiter = node;11     else12         t.nextWaiter = node;13     lastWaiter = node;14     return node;15 }
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行的操作很好理解:

  1. 如果lastWaiter是null,说明FIFO队列中没有任何Node,firstWaiter=Node

  2. 如果lastWaiter不是null,说明FIFO队列中有Node,原lastWaiter的next指向Node

  3. 无论如何,新加入的Node编程lastWaiter,即新加入的Node一定是在最后面

用一张图表示一下构建的数据结构就是:

对比学习,我们总结一下Condition构建出来的队列和AbstractQueuedSynchronizer构建出来的队列的差别,主要体现在2点上:

  1. AbstractQueuedSynchronizer构建出来的队列,头节点是一个没有Thread的空节点,其标识作用,而Condition构建出来的队列,头节点就是真正等待的节点

  2. AbstractQueuedSynchronizer构建出来的队列,节点之间有next与pred相互标识该节点的前一个节点与后一个节点的地址,而Condition构建出来的队列,只使用了nextWaiter标识下一个等待节点的地址

整个过程中,我们看到没有使用任何CAS操作,firstWaiter和lastWaiter也没有用volatile修饰,其实原因很简单:要await()必然要先lock(),既然lock()了就表示没有竞争,没有竞争自然也没必要使用volatile+CAS的机制去保证什么

Condition的await()方法实现原理----线程等待

前面我们看了Condition构建等待队列的过程,接下来我们看一下等待的过程,await()方法的代码比较短,再贴一下:

 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) // clean up if cancelled15         unlinkCancelledWaiters();16     if (interruptMode != 0)17         reportInterruptAfterWait(interruptMode);18 }
Salin selepas log masuk
Salin selepas log masuk

构建完毕队列之后,执行第5行的fullyRelease方法,顾名思义:fullyRelease方法的作用是完全释放Node的状态。方法实现为:

 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 final boolean isOnSyncQueue(Node node) { 2     if (node.waitStatus == Node.CONDITION || node.prev == null) 3         return false; 4     if (node.next != null) // If has successor, it must be on queue 5         return true; 6     /* 7      * node.prev can be non-null, but not yet on queue because 8      * the CAS to place it on queue can fail. So we have to 9      * traverse from tail to make sure it actually made it.  It10      * will always be near the tail in calls to this method, and11      * unless the CAS failed (which is unlikely), it will be12      * there, so we hardly ever traverse much.13      */14     return findNodeFromTail(node);15 }
Salin selepas log masuk

注意这里的判断是Node是否在AbstractQueuedSynchronizer构建的队列中而不是Node是否在Condition构建的队列中,如果Node不在AbstractQueuedSynchronizer构建的队列中,那么调用LockSupport的park方法阻塞。

至此调用await()方法的线程构建Condition等待队列--释放锁--等待的过程已经全部分析完毕。

Condition的signal()实现原理

上面的代码分析了构建Condition等待队列--释放锁--等待的过程,接着看一下signal()方法通知是如何实现的:

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 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行的代码很好理解:

  1. 重新设置firstWaiter,指向第一个waiter的nextWaiter

  2. 如果第一个waiter的nextWaiter为null,说明当前队列中只有一个waiter,lastWaiter置空

  3. 因为firstWaiter是要被signal的,因此它没什么用了,nextWaiter置空

接着执行第6行和第7行的代码,这里重点就是第6行的transferForSignal方法:

 1 final boolean transferForSignal(Node node) { 2     /* 3      * If cannot change waitStatus, the node has been cancelled. 4      */ 5     if (!compareAndSetWaitStatus(node, Node.CONDITION, 0)) 6         return false; 7  8     /* 9      * Splice onto queue and try to set waitStatus of predecessor to10      * indicate that thread is (probably) waiting. If cancelled or11      * attempt to set waitStatus fails, wake up to resync (in which12      * case the waitStatus can be transiently and harmlessly wrong).13      */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 /** 2  * @author 五月的仓颉 3  */ 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 /** 2  * @author 五月的仓颉 3  */ 4 @Test 5 public void testCondition() throws Exception { 6     Lock lock = new ReentrantLock(); 7     Condition condition = lock.newCondition(); 8          9     // 线程0的作用是signal10     Runnable runnable0 = new ConditionThread(lock, condition);11     Thread thread0 = new Thread(runnable0);12     thread0.setName("线程0");13     // 线程1的作用是await14     Runnable runnable1 = new ConditionThread(lock, condition);15     Thread thread1 = new Thread(runnable1);16     thread1.setName("线程1");17     // 线程2的作用是lock18     Runnable runnable2 = new ConditionThread(lock, condition);19     Thread thread2 = new Thread(runnable2);20     thread2.setName("线程2");21         22     thread1.start();23     Thread.sleep(1000);24     thread0.start();25     Thread.sleep(1000);26     thread2.start();27         28     thread1.join();29 }
Salin selepas log masuk

测试代码的意思是:

  1. 线程1先启动,获取锁,调用await()方法等待

  2. 线程0后启动,获取锁,休眠5秒准备signal()

  3. 线程2最后启动,获取锁,由于线程0未使用完毕锁,因此线程2排队,可以此时由于线程0还未signal(),因此线程1在线程0执行signal()后,在AbstractQueuedSynchronizer队列中的顺序是在线程2后面的

代码执行结果为:

<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!

Kenyataan Laman Web ini
Kandungan artikel ini disumbangkan secara sukarela oleh netizen, dan hak cipta adalah milik pengarang asal. Laman web ini tidak memikul tanggungjawab undang-undang yang sepadan. Jika anda menemui sebarang kandungan yang disyaki plagiarisme atau pelanggaran, sila hubungi admin@php.cn
Tutorial Popular
Lagi>
Muat turun terkini
Lagi>
kesan web
Kod sumber laman web
Bahan laman web
Templat hujung hadapan