剖析AbstractQueuedSynchronizer：獨佔模式重難點
- 巴扎黑原創
- 2017-06-23 10:56:241371瀏覽
關於AbstractQueuedSynchronizer
#JDK1.5之後引入了並發包java.util.concurrent，大大提高了Java程式的並發效能。關於java.util.concurrent套件我總結如下：
#AbstractQueuedSynchronizer是並發類別諸如ReentrantLock、CountDownLatch、Semphore的核心
#CAS演算法是AbstractQueuedSynchronizer的核心
可以說AbstractQueuedSynchronizer是並發類別的重中之重。其實之前在ReentrantLock實現原理深入探究一文中已經有結合ReentrantLock詳細解讀AbstractQueuedSynchronizer，但限於當時水平原因，回看一年半前的此文，感覺對於AbstractQueuedSynchronizer的解讀理解還不夠深，因此這裡更新一篇文章，再次解讀AbstractQueuedSynchronizer的資料結構即相關原始碼實現，本文基於JDK1.7版本。
AbstactQueuedSynchronizer的基本資料結構
AbstractQueuedSynchronizer的基本資料結構為Node，關於Node，JDK作者寫了詳細的註釋，這裡我大致總結幾點：
#AbstractQueuedSynchronizer的等待隊列是CLH隊列的變種，CLH隊列通常用於自旋鎖，AbstractQueuedSynchronizer的等待隊列用於阻塞同步器
#每個節點中持有一個名為"status"的字段用於是否一條線程應當阻塞的追踪，但是status字段並不保證加鎖
一條線程如果它處於隊列的頭，那麼他會嘗試去acquire，但是成為頭不保證成功，它只是有權利去競爭
要進入佇列，你只需要自動將它拼接在佇列尾部；要從佇列中移除，你只需要設定header欄位
下面我用一張表格總結Node中持有哪些變數且每個變數的意思：
關於SIGNAL、CANCELLED、CONDITION、PROPAGATE四個狀態，JDK原始碼的註解中同樣有了詳細的解讀，再用一張表格總結一下：
#AbstractQueuedSynchronizer供子類別實作的方法
#AbstractQueuedSynchzonizer是基於模板模式的實現，不過它的模板模式寫法有點特別，整個類別中沒有任何一個abstract的抽象方法，取而代之的是，需要子類別去實現的那些方法通過一個方法體拋出UnsupportedOperationException異常來讓子類別知道。
AbstractQueuedSynchronizer類別中一共有五處方法供子類別實現，用表格總結一下：
這裡的acquire不好翻譯，所以就直接原詞放上來了，因為acquire是動詞，後面並沒有帶賓語，因此不知道具體acquire的是什麼。依照我個人理解，acquire的意思是依照狀態欄位state去取得一個執行目前動作的資格。
例如ReentrantLock的lock()方法最後會呼叫acquire方法，那麼：
線程1去lock()，執行acquire，發現state=0，因此有資格執行lock()的動作，將state設為1，回傳true
執行緒2去lock()，執行acquire，發現state=1，因此沒有資格執行lock()的動作，返回false
這種理解我認為應當是比較準確的。
獨佔模式acquire實作流程
有了上面的這些基礎，我們看一下獨佔式acquire的實作流程，主要是在線程acquire失敗後，是如何建構資料結構的，先看理論，之後再用一個範例畫圖說明。
看看AbstractQuueuedSynchronizer的acquire方法實作流程，acquire方法是用於獨佔模式下進行操作的：
## 1 public final void acquire(int arg) { 2 if (!tryAcquire(arg) && 3 acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) 4 selfInterrupt(); 5 }
tryAcquire方法前面說過了，是子類別實作的一個方法，如果tryAcquire回傳的是true（成功），即表示目前執行緒獲得了一個執行目前動作的資格，自然也就不需要建構資料結構進行阻塞等待。
如果tryAcquire方法回傳的是false，那麼目前執行緒沒有取得執行目前動作的資格，接著執行"acquireQueued(addWaiter(Node.EXCLUSIVE), arg))"這句程式碼，這句話很明顯，它是由兩個步驟構成的：
#addWaiter，加入一個等待者
acquireQueued，試著從等待佇列去取得執行一次acquire動作
分別看一下每一步做了什麼。
addWaiter
#先看第一步，addWaiter做了什麼，從傳入的參數Node.EXCLUSIVE我們知道這是獨佔模式的：
1 private Node addWaiter(Node mode) { 2 Node node = new Node(Thread.currentThread(), mode); 3 // Try the fast path of enq; backup to full enq on failure 4 Node prev = tail; 5 if (prev != null) { 6 node.prev = prev; 7 if (compareAndSetTail(prev, node)) { 8 prev.next = node; 9 return node;10 }11 }12 enq(node);13 return node;14 }
首先看第4行~第11行的程式碼，取得目前資料結構中的尾節點，如果有尾節點，那麼先取得這個節點認為它是前驅節點prev，然後：
- #新產生的Node的前驅節點指向prev ##並發下只有一條執行緒可以透過CAS演算法讓自己的Node成為尾節點，此時將此prev的next指向該執行緒對應的Node
- ##因此
在資料結構中有節點的情況下，所有新增節點都是作為尾節點插入資料結構。從註釋來看，這段邏輯的存在的意義是以最短路徑O(1)的效果完成快速入隊，以最大化減小開銷。
假如目前節點沒有被設定為尾節點，那麼執行enq方法： 1 private Node enq(final Node node) { 2 for (;;) { 3 Node t = tail; 4 if (t == null) { // Must initialize 5 if (compareAndSetHead(new Node())) 6 tail = head; 7 } else { 8 node.prev = t; 9 if (compareAndSetTail(t, node)) {10 t.next = node;11 return t;12 }13 }14 }15 }
這段程式碼的邏輯為：
如果尾節點為空，即當前資料結構中沒有節點，那麼new一個不帶任何狀態的Node作為頭節點
#如果尾節點不為空，那麼並發下使用CAS演算法將目前Node追加成為尾節點，由於是for(;;)循環，因此所有沒有成功acquire的Node最後都會被追加到資料結構中
就看完了程式碼，用一張圖表示AbstractQueuedSynchronizer的整體資料結構（比較簡單，就不自己畫了，網路上隨便找了一張圖）：
acquireQueued
佇列建構好了，下一步就是在必要的時候從佇列裡面拿出一個Node了，這就是acquireQueued方法，顧名思義，從佇列裡面acquire。看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.prevecessor(); 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 }
这段代码描述了几件事：
从第6行的代码获取节点的前驱节点p，第7行的代码判断p是前驱节点并tryAcquire我们知道，只有当前第一个持有Thread的节点才会尝试acquire，如果节点acquire成功，那么setHead方法，将当前节点作为head、将当前节点中的thread设置为null、将当前节点的prev设置为null，这保证了数据结构中头结点永远是一个不带Thread的空节点
如果当前节点不是前驱节点或者tryAcquire失败，那么执行第13行~第15行的代码，做了两步操作，首先判断在acquie失败后是否应该park，其次park并检查中断状态
看一下第一步shouldParkAfterFailedAcquire代码做了什么：
1 private static boolean shouldParkAfterFailedAcquire(Node prev, Node node) { 2 int ws = prev.waitStatus; 3 if (ws == Node.SIGNAL) 4 /* 5 * This node has already set status asking a release 6 * to signal it, so it can safely park. 7 */ 8 return true; 9 if (ws > 0) {10 /*11 * prevecessor was cancelled. Skip over prevecessors and12 * indicate retry.13 */14 do {15 node.prev = prev = prev.prev;16 } while (prev.waitStatus > 0);17 prev.next = node;18 } else {19 /*20 * waitStatus must be 0 or PROPAGATE. Indicate that we21 * need a signal, but don't park yet. Caller will need to22 * retry to make sure it cannot acquire before parking.23 */24 compareAndSetWaitStatus(prev, ws, Node.SIGNAL);25 }26 return false;27 }这里每个节点判断它前驱节点的状态，如果：
它的前驱节点是SIGNAL状态的，返回true，表示当前节点应当park
它的前驱节点的waitStatus>0，相当于CANCELLED（因为状态值里面只有CANCELLED是大于0的），那么CANCELLED的节点作废，当前节点不断向前找并重新连接为双向队列，直到找到一个前驱节点waitStats不是CANCELLED的为止
它的前驱节点不是SIGNAL状态且waitStatus<=0，此时执行第24行代码，利用CAS机制，如果waitStatus的前驱节点是0那么更新为SIGNAL状态
如果判断判断应当park，那么parkAndCheckInterrupt方法：
LockSupport.park(
利用LockSupport的park方法让当前线程阻塞。
独占模式release流程
上面整理了独占模式的acquire流程，看到了等待的Node是如何构建成一个数据结构的，下面看一下释放的时候做了什么，release方法的实现为：
1 public final boolean release(int arg) {2 if (tryRelease(arg)) {3 Node h = head;4 if (h != null && h.waitStatus != 0)5 unparkSuccessor(h);6 return true;7 }8 return false;9 }tryRelease同样是子类去实现的，表示当前动作我执行完了，要释放我执行当前动作的资格，讲这个资格让给其它线程，然后tryRelease释放成功，获取到head节点，如果head节点的waitStatus不为0的话，执行unparkSuccessor方法，顾名思义unparkSuccessor意为unpark头结点的继承者，方法实现为：
1 private void unparkSuccessor(Node node) { 2 /* 3 * If status is negative (i.e., possibly needing signal) try 4 * to clear in anticipation of signalling. It is OK if this 5 * fails or if status is changed by waiting thread. 6 */ 7 int ws = node.waitStatus; 8 if (ws < 0) 9 compareAndSetWaitStatus(node, ws, 0);10 11 /*12 * Thread to unpark is held in successor, which is normally13 * just the next node. But if cancelled or apparently null,14 * traverse backwards from tail to find the actual15 * non-cancelled successor.16 */17 Node s = node.next;18 if (s == null || s.waitStatus > 0) {19 s = null;20 for (Node t = tail; t != null && t != node; t = t.prev)21 if (t.waitStatus <= 0)22 s = t;23 }24 if (s != null)25 LockSupport.unpark(s.thread);26 }这段代码比较好理解，整理一下流程：
头节点的waitStatus<0，将头节点的waitStatus设置为0
拿到头节点的下一个节点s，如果s==null或者s的waitStatus>0（被取消了），那么从队列尾巴开始向前寻找一个waitStatus<=0的节点作为后继要唤醒的节点
最后，如果拿到了一个不等于null的节点s，就利用LockSupport的unpark方法让它取消阻塞。
实战举例：数据结构构建
上面的例子讲解地过于理论，下面利用ReentrantLock举个例子，但是这里不讲ReentrantLock实现原理，只是利用ReentrantLock研究AbstractQueuedSynchronizer的acquire和release。示例代码为：
1 /** 2 * @author 五月的仓颉 3 */ 4 public class AbstractQueuedSynchronizerTest { 5 6 @Test 7 public void testAbstractQueuedSynchronizer() { 8 Lock lock = new ReentrantLock(); 9 10 Runnable runnable0 = new ReentrantLockThread(lock);11 Thread thread0 = new Thread(runnable0);12 thread0.setName("线程0");13 14 Runnable runnable1 = new ReentrantLockThread(lock);15 Thread thread1 = new Thread(runnable1);16 thread1.setName("线程1");17 18 Runnable runnable2 = new ReentrantLockThread(lock);19 Thread thread2 = new Thread(runnable2);20 thread2.setName("线程2");21 22 thread0.start();23 thread1.start();24 thread2.start();25 26 for (;;);27 }28 29 private class ReentrantLockThread implements Runnable {30 31 private Lock lock;32 33 public ReentrantLockThread(Lock lock) {34 this.lock = lock;35 }36 37 @Override38 public void run() {39 try {40 lock.lock();41 for (;;);42 } finally {43 lock.unlock();44 }45 }46 47 }48 49 }全部是死循环，相当于第一条线程（线程0）acquire成功之后，后两条线程（线程1、线程2）阻塞，下面的代码就不考虑后两条线程谁先谁后的问题，就一条线程（线程1）流程执行到底、另一条线程（线程2）流程执行到底这么分析了。
这里再把addWaiter和enq两个方法源码贴一下：
1 private Node addWaiter(Node mode) { 2 Node node = new Node(Thread.currentThread(), mode); 3 // Try the fast path of enq; backup to full enq on failure 4 Node prev = tail; 5 if (prev != null) { 6 node.prev = prev; 7 if (compareAndSetTail(prev, node)) { 8 prev.next = node; 9 return node;10 }11 }12 enq(node);13 return node;14 } 1 private Node enq(final Node node) { 2 for (;;) { 3 Node t = tail; 4 if (t == null) { // Must initialize 5 if (compareAndSetHead(new Node())) 6 tail = head; 7 } else { 8 node.prev = t; 9 if (compareAndSetTail(t, node)) {10 t.next = node;11 return t;12 }13 }14 }15 }首先第一个acquire失败的线程1，由于此时整个数据结构中么没有任何数据，因此addWaiter方法第4行中拿到的prev=tail为空，执行enq方法，首先第3行获取tail，第4行判断到tail是null，因此头结点new一个Node出来通过CAS算法设置为数据结构的head，tail同样也是这个Node，此时数据结构为：
为了方便描述，prev和next，我给每个Node随便加了一个地址。接着继续enq，因为enq内是一个死循环，所以继续第3行获取tail，new了一个空的Node之后tail就有了，执行else判断，通过第8行~第10行代码将当前线程对应的Node追加到数据结构尾部，那么当前构建的数据结构为：
这样，线程1对应的Node被加入数据结构，成为数据结构的tail，而数据结构的head是一个什么都没有的空Node。
接着线程2也acquire失败了，线程2既然acquire失败，那也要准备被加入数据结构中，继续先执行addWaiter方法，由于此时已经有了tail，因此不需要执行enq方法，可以直接将当前Node添加到数据结构尾部，那么当前构建的数据结构为：
至此，两个阻塞的线程构建的三个Node已经全部归位。
实战举例：线程阻塞
上述流程只是描述了构建数据结构的过程，并没有描述线程1、线程2阻塞的流程，因此接着继续用实际例子看一下线程1、线程2如何阻塞。贴一下acquireQueued、shouldParkAfterFailedAcquire两个方法源码：
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.prevecessor(); 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 } 1 private static boolean shouldParkAfterFailedAcquire(Node prev, Node node) { 2 int ws = prev.waitStatus; 3 if (ws == Node.SIGNAL) 4 /* 5 * This node has already set status asking a release 6 * to signal it, so it can safely park. 7 */ 8 return true; 9 if (ws > 0) {10 /*11 * prevecessor was cancelled. Skip over prevecessors and12 * indicate retry.13 */14 do {15 node.prev = prev = prev.prev;16 } while (prev.waitStatus > 0);17 prev.next = node;18 } else {19 /*20 * waitStatus must be 0 or PROPAGATE. Indicate that we21 * need a signal, but don't park yet. Caller will need to22 * retry to make sure it cannot acquire before parking.23 */24 compareAndSetWaitStatus(prev, ws, Node.SIGNAL);25 }26 return false;27 }</span></p></li>
</ol>
</div>
<p><span style="font-size: 13px; font-family: 宋体">首先是线程1，它的前驱节点是head节点，在它tryAcquire成功的情况下，执行第8行~第11行的代码。做几件事情：</span></p>
<ol class=" list-paddingleft-2">
<li><p><span style="font-size: 13px; font-family: 宋体">head为线程1对应的Node</span></p></li>
<li><p><span style="font-size: 13px; font-family: 宋体">线程1对应的Node的thread置空</span></p></li>
<li><p><span style="font-size: 13px; font-family: 宋体">线程1对应的Node的prev置空</span></p></li>
<li><p><span style="font-size: 13px; font-family: 宋体">原head的next置空，这样原head中的prev、next、thread都为空，对象内没有引用指向其他地方，GC可以认为这个Node是垃圾，对这个Node进行回收，注释"Help GC"就是这个意思</span></p></li>
<li><p><span style="font-size: 13px; font-family: 宋体">failed=false表示没有失败</span></p></li>
</ol>
<p><span style="font-size: 13px; font-family: 宋体">因此，如果线程1执行tryAcquire成功，那么数据结构将变为：</span></p>
<p><span style="font-size: 13px; font-family: 宋体"><img src="https://img.php.cn/upload/article/000/000/001/3d801230fc075de2d75122d27258c2ad-7.png" alt=""></span></p>
<p><span style="font-size: 13px; font-family: 宋体">从上述流程可以总结到：<span style="color: #ff0000"><strong>只有前驱节点为head的节点会尝试tryAcquire，其余都不会，结合后面的release选继承者的方式，保证了先acquire失败的线程会优先从阻塞状态中解除去重新acquire</strong></span>。这是一种公平的acquire方式，因为它遵循"先到先得"原则，但是我们可以动动手脚让这种公平变为非公平，比如ReentrantLock默认的费公平模式，这个留在后面说。</span></p>
<p><span style="font-size: 13px; font-family: 宋体">那如果线程1执行tryAcquire失败，那么要执行shouldParkAfterFailedAcquire方法了，shouldParkAfterFailedAcquire拿线程1的前驱节点也就是head节点的waitStatus做了一个判断，因为waitStatus=0，因此执行第18行~第20行的逻辑，将head的waitStatus设置为SIGNAL即-1，然后方法返回false，数据结构变为：</span></p>
<p><span style="font-size: 13px; font-family: 宋体"><img src="https://img.php.cn/upload/article/000/000/001/3d801230fc075de2d75122d27258c2ad-8.png" alt=""></span></p>
<p><span style="font-size: 13px; font-family: 宋体">看到这里就一个变化：head的waitStatus从0变成了-1。既然shouldParkAfterFailedAcquire返回false，acquireQueued的第13行~第14行的判断自然不通过，继续走for(;;)循环，如果tryAcquire失败显然又来到了shouldParkAfterFailedAcquire方法，此时线程1对应的Node的前驱节点head节点的waitStatus已经变为了SIGNAL即-1，因此执行第4行~第8行的代码，直接返回true出去。</span></p>
<p><span style="font-size: 13px; font-family: 宋体">shouldParkAfterFailedAcquire返回true，parkAndCheckInterrupt直接调用LockSupport的park方法：</span></p>
<div class="cnblogs_code"><pre class="brush:php;toolbar:false"> 1 private final boolean parkAndCheckInterrupt() { 2 LockSupport.park(this); 3 return Thread.interrupted(); 4 }至此线程1阻塞，线程2阻塞的流程与线程1阻塞的流程相同，可以自己分析一下。
另外再提一个问题，不知道大家会不会想：
为什么线程1对应的Node构建完毕不直接调用LockSupport的park方法进行阻塞？
为什么不直接把head的waitStatus直接设置为Signal而要从0设置为Signal？
我认为这是AbstractQueuedSynchronizer开发人员做了类似自旋的操作。因为很多时候获取acquire进行操作的时间很短，阻塞会引起上下文的切换，而很短时间就从阻塞状态解除，这样相对会比较耗费性能。
因此我们看到线程1自构建完毕Node加入数据结构到阻塞，一共尝试了两次tryAcquire，如果其中有一次成功，那么线程1就没有必要被阻塞，提升了性能。
以上是剖析AbstractQueuedSynchronizer：獨佔模式重難點的詳細內容。更多資訊請關注PHP中文網其他相關文章！