如何理解Java并發(fā)里的Condition

在java.util.concurrent包中，有兩個很特殊的工具類，Condition和ReentrantLock，使用過的人都知道，ReentrantLock（重入鎖）是jdk的concurrent包提供的一種獨占鎖的實現。它繼承自Dong Lea的 AbstractQueuedSynchronizer（同步器），確切的說是ReentrantLock的一個內部類繼承了AbstractQueuedSynchronizer，ReentrantLock只不過是代理了該類的一些方法，可能有人會問為什么要使用內部類在包裝一層？ 我想是安全的關系，因為AbstractQueuedSynchronizer中有很多方法，還實現了共享鎖，Condition(稍候再細說)等功能，如果直接使ReentrantLock繼承它，則很容易出現AbstractQueuedSynchronizer中的API被無用的情況。

言歸正傳，今天，我們討論下Condition工具類的實現。

ReentrantLock和Condition的使用方式通常是這樣的：

 public static void main(String[] args) { final ReentrantLock reentrantLock = new ReentrantLock(); final Condition condition = reentrantLock.newCondition(); Thread thread = new Thread((Runnable) () -> { try { reentrantLock.lock(); System.out.println("我要等一個新信號" + this); condition.wait(); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("拿到一個信號??！" + this); reentrantLock.unlock(); }, "waitThread1"); thread.start(); Thread thread1 = new Thread((Runnable) () -> { reentrantLock.lock(); System.out.println("我拿到鎖了"); try { Thread.sleep(3000); } catch (InterruptedException e) { e.printStackTrace(); } condition.signalAll(); System.out.println("我發(fā)了一個信號??！"); reentrantLock.unlock(); }, "signalThread"); thread1.start(); }

運行后，結果如下：

我要等一個新信號lock.ReentrantLockTest$1@a62fc3我拿到鎖了我發(fā)了一個信號??！拿到一個信號?。?/pre>
可以看到，
Condition的執(zhí)行方式，是當在線程1中調用await方法后，線程1將釋放鎖，并且將自己沉睡，等待喚醒，
線程2獲取到鎖后，開始做事，完畢后，調用Condition的signal方法，喚醒線程1，線程1恢復執(zhí)行。
以上說明Condition是一個多線程間協調通信的工具類，使得某個，或者某些線程一起等待某個條件（Condition）,只有當該條件具備( signal 或者 signalAll方法被帶調用)時 ，這些等待線程才會被喚醒，從而重新爭奪鎖。
那，它是怎么實現的呢？
首先還是要明白，reentrantLock.newCondition() 返回的是Condition的一個實現，該類在AbstractQueuedSynchronizer中被實現，叫做newCondition()
public Condition newCondition() { return sync.newCondition(); }
它可以訪問AbstractQueuedSynchronizer中的方法和其余內部類（AbstractQueuedSynchronizer是個抽象類，至于他怎么能訪問，這里有個很奇妙的點，后面我專門用demo說明 ）
現在，我們一起來看下Condition類的實現，還是從上面的demo入手，
為了方便書寫，我將AbstractQueuedSynchronizer縮寫為AQS
當await被調用時，代碼如下：
 public final void await() throws InterruptedException { if (Thread.interrupted()) throw new InterruptedException(); Node node = addConditionWaiter(); // 將當前線程包裝下后， // 添加到Condition自己維護的一個鏈表中。 int savedState = fullyRelease(node);// 釋放當前線程占有的鎖，從demo中看到， // 調用await前，當前線程是占有鎖的 int interruptMode = 0; while (!isOnSyncQueue(node)) {// 釋放完畢后，遍歷AQS的隊列，看當前節(jié)點是否在隊列中， // 不在 說明它還沒有競爭鎖的資格，所以繼續(xù)將自己沉睡。 // 直到它被加入到隊列中，聰明的你可能猜到了， // 沒有錯，在singal的時候加入不就可以了？ 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); }
回到上面的demo，鎖被釋放后，線程1開始沉睡，這個時候線程因為線程1沉睡時，會喚醒AQS隊列中的頭結點，所所以線程2會開始競爭鎖，并獲取到，等待3秒后，線程2會調用signal方法，"發(fā)出"signal信號，signal方法如下：
 public final void signal() { if (!isHeldExclusively()) throw new IllegalMonitorStateException(); Node first = firstWaiter; // firstWaiter為condition自己維護的一個鏈表的頭結點， // 取出第一個節(jié)點后開始喚醒操作 if (first != null) doSignal(first); }
說明下，其實Condition內部維護了等待隊列的頭結點和尾節(jié)點，該隊列的作用是存放等待signal信號的線程，該線程被封裝為Node節(jié)點后存放于此。
 public class ConditionObject implements Condition, java.io.Serializable { private static final long serialVersionUID = 1173984872572414699L; /** First node of condition queue. */ private transient Node firstWaiter; /** Last node of condition queue. */ private transient Node lastWaiter;
關鍵的就在于此，我們知道AQS自己維護的隊列是當前等待資源的隊列，AQS會在資源被釋放后，依次喚醒隊列中從前到后的所有節(jié)點，使他們對應的線程恢復執(zhí)行。直到隊列為空。
而Condition自己也維護了一個隊列，該隊列的作用是維護一個等待signal信號的隊列，兩個隊列的作用是不同，事實上，每個線程也僅僅會同時存在以上兩個隊列中的一個，流程是這樣的：
線程1調用reentrantLock.lock時，線程被加入到AQS的等待隊列中。
線程1調用await方法被調用時，該線程從AQS中移除，對應操作是鎖的釋放。
接著馬上被加入到Condition的等待隊列中，以為著該線程需要signal信號。
線程2，因為線程1釋放鎖的關系，被喚醒，并判斷可以獲取鎖，于是線程2獲取鎖，并被加入到AQS的等待隊列中。
線程2調用signal方法，這個時候Condition的等待隊列中只有線程1一個節(jié)點，于是它被取出來，并被加入到AQS的等待隊列中。 注意，這個時候，線程1 并沒有被喚醒。
signal方法執(zhí)行完畢，線程2調用reentrantLock.unLock()方法，釋放鎖。這個時候因為AQS中只有線程1，于是，AQS釋放鎖后按從頭到尾的順序喚醒線程時，線程1被喚醒，于是線程1回復執(zhí)行。
直到釋放所整個過程執(zhí)行完畢。
可以看到，整個協作過程是靠結點在AQS的等待隊列和Condition的等待隊列中來回移動實現的，Condition作為一個條件類，很好的自己維護了一個等待信號的隊列，并在適時的時候將結點加入到AQS的等待隊列中來實現的喚醒操作。
看到這里，signal方法的代碼應該不難理解了。
取出頭結點，然后doSignal
 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) && // 將老的頭結點，加入到AQS的等待隊列中 (first = firstWaiter) != null); } final boolean transferForSignal(Node node) { /* * If cannot change waitStatus, the node has been cancelled. */ if (!compareAndSetWaitStatus(node, Node.CONDITION, 0)) return false; /* * Splice onto queue and try to set waitStatus of predecessor to * indicate that thread is (probably) waiting. If cancelled or attempt * to set waitStatus fails, wake up to resync (in which case the * waitStatus can be transiently and harmlessly wrong). */ Node p = enq(node); int ws = p.waitStatus; // 如果該結點的狀態(tài)為cancel 或者修改waitStatus失敗，則直接喚醒。 if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL)) LockSupport.unpark(node.thread); return true; }
可以看到，正常情況 ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL)這個判斷是不會為true的，所以，不會在這個時候喚醒該線程。
只有到發(fā)送signal信號的線程調用reentrantLock.unlock()后因為它已經被加到AQS的等待隊列中，所以才會被喚醒。
            

                標題名稱：如何理解Java并發(fā)里的Condition
                

                鏈接URL：http://www.bm7419.com/news40/100090.html
            
            

                                    成都網站建設公司_創(chuàng)新互聯，為您提供網頁設計公司、網站策劃、自適應網站、ChatGPT、虛擬主機、搜索引擎優(yōu)化            
            

                廣告
                
            
            

                聲明：本網站發(fā)布的內容（圖片、視頻和文字）以用戶投稿、用戶轉載內容為主，如果涉及侵權請盡快告知，我們將會在第一時間刪除。文章觀點不代表本網站立場，如需處理請聯系客服。電話：028-86922220；郵箱：631063699@qq.com。內容未經允許不得轉載，或轉載時需注明來源：
                創(chuàng)新互聯