Java併發編程-看懂AQS的前世今生

　　在具備了volatile、CAS和模板方法設計模式的知識之後，我們可以來深入學習下AbstractQueuedSynchronizer（AQS），本文主要想從AQS的產生背景、設計和結構、源代碼實現及AQS應用這4個方面來學習下AQS，文章耗時一個月，所以篇幅有點長，需要一點耐心。

　　1、AQS產生背景

　　通過JCP的JSR166規範，Jdk1.5開始引入了j.u.c包，這個包提供了一系列支持併發的組件。這些組件是一系列的同步器，這些同步器主要維護著以下幾個功能：內部同步狀態的管理(例如表示一個鎖的狀態是獲取還是釋放)，同步狀態的更新和檢查操作，且至少有一個方法會導致調用線程在同步狀態被獲取時阻塞，以及在其他線程改變這個同步狀態時解除線程的阻塞。上述的這些的實際例子包括：互斥排它鎖的不同形式、讀寫鎖、信號量、屏障、Future、事件指示器以及傳送隊列等。可以看下這裡的4.2的圖便能理解j.u.c包的組件構成。

　　幾乎任一同步器都可以用來實現其他形式的同步器。例如，可以用可重入鎖實現信號量或者用信號量實現可重入鎖。但是，這樣做帶來的複雜性、開銷及不靈活使j.u.c最多只能是一個二流工程，且缺乏吸引力。如果任何這樣的構造方式不能在本質上比其他形式更簡潔，那麼開發者就不應該隨意地選擇其中的某個來構建另一個同步器。因此，JSR166基於AQS類建立了一個小框架，這個框架為構造同步器提供一種通用的機制，並且被j.u.c包中大部分類使用，同時很多用戶也可以用它來定義自己的同步器。這個就是j.u.c的作者Doug Lea大神的初衷，通過提供AQS這個基礎組件來構建j.u.c的各種工具類，至此就可以理解AQS的產生背景了。

　　2、AQS的設計和結構

　　2.1 設計思想

　　同步器的核心方法是acquire和release操作，其背後的思想也比較簡潔明確。acquire操作是這樣的：

　　while (當前同步器的狀態不允許獲取操作) {

        　　如果當前線程不在隊列中，則將其插入隊列

        　　阻塞當前線程

　　}

　　如果線程位於隊列中，則將其移出隊列

　　release操作是這樣的：

　　更新同步器的狀態

　　if (新的狀態允許某個被阻塞的線程獲取成功)

         　　解除隊列中一個或多個線程的阻塞狀態

　　從這兩個操作中的思想中我們可以提取出三大關鍵操作：同步器的狀態變更、線程阻塞和釋放、插入和移出隊列。所以為了實現這兩個操作，需要協調三大關鍵操作引申出來的三個基本組件：

　　·同步器狀態的原子性管理；

　　·線程阻塞與解除阻塞；

　　·隊列的管理；

　　由這三個基本組件，我們來看j.u.c是怎麼設計的。

　　2.1.1 同步狀態

　　AQS類使用單個int（32位）來保存同步狀態，並暴露出getState、setState以及compareAndSet操作來讀取和更新這個同步狀態。其中屬性state被聲明為volatile，並且通過使用CAS指令來實現compareAndSetState，使得當且僅當同步狀態擁有一個一致的期望值的時候，才會被原子地設置成新值，這樣就達到了同步狀態的原子性管理，確保了同步狀態的原子性、可見性和有序性。

 　　基於AQS的具體實現類（如鎖、信號量等）必鬚根據暴露出的狀態相關的方法定義tryAcquire和tryRelease方法，以控制acquire和release操作。當同步狀態滿足時，tryAcquire方法必須返回true，而當新的同步狀態允許後續acquire時，tryRelease方法也必須返回true。這些方法都接受一個int類型的參數用於傳遞想要的狀態。

　　2.1.2 阻塞

　　直到JSR166，阻塞線程和解除線程阻塞都是基於Java的內置管程，沒有其它非基於Java內置管程的API可以用來達到阻塞線程和解除線程阻塞。唯一可以選擇的是Thread.suspend和Thread.resume，但是它們都有無法解決的競態問題，所以也沒法用，目前該方法基本已被拋棄。具體不能用的原因可以官方給出的答覆。

　　j.u.c.locks包提供了LockSupport類來解決這個問題。方法LockSupport.park阻塞當前線程直到有個LockSupport.unpark方法被調用。unpark的調用是沒有被計數的，因此在一個park調用前多次調用unpark方法只會解除一個park操作。另外，它們作用於每個線程而不是每個同步器。一個線程在一個新的同步器上調用park操作可能會立即返回，因為在此之前可以有多餘的unpark操作。但是，在缺少一個unpark操作時，下一次調用park就會阻塞。雖然可以顯式地取消多餘的unpark調用，但並不值得這樣做。在需要的時候多次調用park會更高效。park方法同樣支持可選的相對或絕對的超時設置，以及與JVM的Thread.interrupt結合 ，可通過中斷來unpark一個線程。

　　2.1.3 隊列

　　整個框架的核心就是如何管理線程阻塞隊列，該隊列是嚴格的FIFO隊列，因此不支持線程優先順序的同步。同步隊列的最佳選擇是自身沒有使用底層鎖來構造的非阻塞數據結構，業界主要有兩種選擇，一種是MCS鎖，另一種是CLH鎖。其中CLH一般用於自旋，但是相比MCS，CLH更容易實現取消和超時，所以同步隊列選擇了CLH作為實現的基礎。

　　CLH隊列實際並不那麼像隊列，它的出隊和入隊與實際的業務使用場景密切相關。它是一個鏈表隊列，通過AQS的兩個欄位head（頭節點）和tail（尾節點）來存取，這兩個欄位是volatile類型，初始化的時候都指向了一個空節點。如下圖：

　　入隊操作：CLH隊列是FIFO隊列，故新的節點到來的時候，是要插入到當前隊列的尾節點之後。試想一下，當一個線程成功地獲取了同步狀態，其他線程將無法獲取到同步狀態，轉而被構造成為節點並加入到同步隊列中，而這個加入隊列的過程必須要保證線程安全，因此同步器提供了一個CAS方法，它需要傳遞當前線程“認為”的尾節點和當前節點，只有設置成功後，當前節點才正式與之前的尾節點建立關聯。入隊操作示意圖大致如下：

　　出隊操作：因為遵循FIFO規則，所以能成功獲取到AQS同步狀態的必定是首節點，首節點的線程在釋放同步狀態時，會喚醒後續節點，而後續節點會在獲取AQS同步狀態成功的時候將自己設置為首節點。設置首節點是由獲取同步成功的線程來完成的，由於只能有一個線程可以獲取到同步狀態，所以設置首節點的方法不需要像入隊這樣的CAS操作，只需要將首節點設置為原首節點的後續節點同時斷開原節點、後續節點的引用即可。出隊操作示意圖大致如下：

　　這一小節只是簡單的描述了隊列的大概，目的是為了表達清楚隊列的設計框架，實際上CLH隊列已經和初始的CLH隊列已經發生了一些變化，具體的可以看查看資料中Doug Lea的那篇論文中的3.3 Queues。

　　2.1.4 條件隊列

　　上一節的隊列其實是AQS的同步隊列，這一節的隊列是條件隊列，隊列的管理除了有同步隊列，還有條件隊列。AQS只有一個同步隊列，但是可以有多個條件隊列。AQS框架提供了一個ConditionObject類，給維護獨占同步的類以及實現Lock介面的類使用。

　　ConditionObject類實現了Condition介面，Condition介面提供了類似Object管程式的方法，如await、signal和signalAll操作，還擴展了帶有超時、檢測和監控的方法。ConditionObject類有效地將條件與其它同步操作結合到了一起。該類只支持Java風格的管程訪問規則，這些規則中，當且僅噹噹前線程持有鎖且要操作的條件（condition）屬於該鎖時，條件操作才是合法的。這樣，一個ConditionObject關聯到一個ReentrantLock上就表現的跟內置的管程（通過Object.wait等）一樣了。兩者的不同僅僅在於方法的名稱、額外的功能以及用戶可以為每個鎖聲明多個條件。

　　ConditionObject類和AQS共用了內部節點，有自己單獨的條件隊列。signal操作是通過將節點從條件隊列轉移到同步隊列中來實現的，沒有必要在需要喚醒的線程重新獲取到鎖之前將其喚醒。signal操作大致示意圖如下：

　　await操作就是當前線程節點從同步隊列進入條件隊列進行等待，大致示意圖如下：

　　實現這些操作主要複雜在，因超時或Thread.interrupt導致取消了條件等待時，該如何處理。await和signal幾乎同時發生就會有競態問題，最終的結果遵照內置管程相關的規範。JSR133修訂以後，就要求如果中斷發生在signal操作之前，await方法必須在重新獲取到鎖後，拋出InterruptedException。但是，如果中斷發生在signal後，await必須返回且不拋異常，同時設置線程的中斷狀態。

　　2.2 方法結構

　　如果我們理解了上一節的設計思路，我們大致就能知道AQS的主要數據結構了。

　　進而再來看下AQS的主要方法及其作用。

　　看到這，我們對AQS的數據結構應該基本上有一個大致的認識，有了這個基本面的認識，我們就可以來看下AQS的源代碼。

　　3、AQS的源代碼實現

　　主要通過獨占式同步狀態的獲取和釋放、共用式同步狀態的獲取和釋放來看下AQS是如何實現的。

　　3.1 獨占式同步狀態的獲取和釋放

　　獨占式同步狀態調用的方法是acquire，代碼如下：

public final void acquire(int arg) {
        if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }

　　上述代碼主要完成了同步狀態獲取、節點構造、加入同步隊列以及在同步隊列中自旋等待的相關工作，其主要邏輯是：首先調用子類實現的tryAcquire方法，該方法保證線程安全的獲取同步狀態，如果同步狀態獲取失敗，則構造獨占式同步節點（同一時刻只能有一個線程成功獲取同步狀態）並通過addWaiter方法將該節點加入到同步隊列的尾部，最後調用acquireQueued方法，使得該節點以自旋的方式獲取同步狀態。如果獲取不到則阻塞節點中的線程，而被阻塞線程的喚醒主要依靠前驅節點的出隊或阻塞線程被中斷來實現。

　　下麵來首先來看下節點構造和加入同步隊列是如何實現的。代碼如下：

private Node addWaiter(Node mode) {
        // 當前線程構造成Node節點
        Node node = new Node(Thread.currentThread(), mode);
        // Try the fast path of enq; backup to full enq on failure
        // 嘗試快速在尾節點後新增節點 提升演算法效率 先將尾節點指向pred
        Node pred = tail;
        if (pred != null) {
            //尾節點不為空  當前線程節點的前驅節點指向尾節點
            node.prev = pred;
            //併發處理 尾節點有可能已經不是之前的節點 所以需要CAS更新
            if (compareAndSetTail(pred, node)) {
                //CAS更新成功 當前線程為尾節點 原先尾節點的後續節點就是當前節點
                pred.next = node;
                return node;
            }
        }
        //第一個入隊的節點或者是尾節點後續節點新增失敗時進入enq
        enq(node);
        return node;
    }
private Node enq(final Node node) {
        for (;;) {
            Node t = tail;
            if (t == null) { // Must initialize
                //尾節點為空  第一次入隊  設置頭尾節點一致 同步隊列的初始化
                if (compareAndSetHead(new Node()))
                    tail = head;
            } else {
                //所有的線程節點在構造完成第一個節點後 依次加入到同步隊列中
                node.prev = t;
                if (compareAndSetTail(t, node)) {
                    t.next = node;
                    return t;
                }
            }
        }
    }

　　節點進入同步隊列之後，就進入了一個自旋的過程，每個線程節點都在自省地觀察，當條件滿足，獲取到了同步狀態，就可以從這個自旋過程中退出，否則依舊留在這個自旋過程中並會阻塞節點的線程，代碼如下：

final boolean acquireQueued(final Node node, int arg) {
        boolean failed = true;
        try {
            boolean interrupted = false;
            for (;;) {
                //獲取當前線程節點的前驅節點
                final Node p = node.predecessor();
                //前驅節點為頭節點且成功獲取同步狀態
                if (p == head && tryAcquire(arg)) {
                    //設置當前節點為頭節點
                    setHead(node);
                    p.next = null; // help GC
                    failed = false;
                    return interrupted;
                }
                //是否阻塞
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

　　再來看看shouldParkAfterFailedAcquire和parkAndCheckInterrupt是怎麼來阻塞當前線程的，代碼如下：

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        //前驅節點的狀態決定後續節點的行為
　　　　 int ws = pred.waitStatus;
        if (ws == Node.SIGNAL)
            /*前驅節點為-1 後續節點可以被阻塞
             * This node has already set status asking a release
             * to signal it, so it can safely park.
             */
            return true;
        if (ws > 0) {
            /*
             * Predecessor was cancelled. Skip over predecessors and
             * indicate retry.
             */
            do {
                node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0);
            pred.next = node;
        } else {
            /*前驅節點是初始或者共用狀態就設置為-1 使後續節點阻塞
             * waitStatus must be 0 or PROPAGATE.  Indicate that we
             * need a signal, but don't park yet.  Caller will need to
             * retry to make sure it cannot acquire before parking.
             */
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        return false;
    }
private final boolean parkAndCheckInterrupt() {
        //阻塞線程
        LockSupport.park(this);
        return Thread.interrupted();
    }

　　節點自旋的過程大致示意圖如下，其實就是對圖二、圖三的補充。

　　　　　　　　　　　　　　　　圖六  節點自旋獲取隊列同步狀態

　　整個獨占式獲取同步狀態的流程圖大致如下：

　　　　　　　　　　　　圖七  獨占式獲取同步狀態

　　當同步狀態獲取成功之後，當前線程從acquire方法返回，對於鎖這種併發組件而言，就意味著當前線程獲取了鎖。有獲取同步狀態的方法，就存在其對應的釋放方法，該方法為release，現在來看下這個方法的實現，代碼如下：

public final boolean release(int arg) {
        if (tryRelease(arg)) {//同步狀態釋放成功
            Node h = head;
            if (h != null && h.waitStatus != 0)
                //直接釋放頭節點
                unparkSuccessor(h);
            return true;
        }
        return false;
    }
private void unparkSuccessor(Node node) {
        /*
         * If status is negative (i.e., possibly needing signal) try
         * to clear in anticipation of signalling.  It is OK if this
         * fails or if status is changed by waiting thread.
         */
        int ws = node.waitStatus;
        if (ws < 0)
            compareAndSetWaitStatus(node, ws, 0);

        /*尋找符合條件的後續節點
         * Thread to unpark is held in successor, which is normally
         * just the next node.  But if cancelled or apparently null,
         * traverse backwards from tail to find the actual
         * non-cancelled successor.
         */
        Node s = node.next;
        if (s == null || s.waitStatus > 0) {
            s = null;
            for (Node t = tail; t != null && t != node; t = t.prev)
                if (t.waitStatus <= 0)
                    s = t;
        }
        if (s != null)
            //喚醒後續節點
            LockSupport.unpark(s.thread);
    }

　　獨占式釋放是非常簡單而且明確的。

　　總結下獨占式同步狀態的獲取和釋放：在獲取同步狀態時，同步器維護一個同步隊列，獲取狀態失敗的線程都會被加入到隊列中併在隊列中進行自旋；移出隊列的條件是前驅節點為頭節點且成功獲取了同步狀態。在釋放同步狀態時，同步器調用tryRelease方法釋放同步狀態，然後喚醒頭節點的後繼節點。

　　3.2 共用式同步狀態的獲取和釋放

　　共用式同步狀態調用的方法是acquireShared，代碼如下：

public final void acquireShared(int arg) {
        //獲取同步狀態的返回值大於等於0時表示可以獲取同步狀態
        //小於0時表示可以獲取不到同步狀態  需要進入隊列等待
        if (tryAcquireShared(arg) < 0)
            doAcquireShared(arg);
    }
private void doAcquireShared(int arg) {
        //和獨占式一樣的入隊操作
        final Node node = addWaiter(Node.SHARED);
        boolean failed = true;
        try {
            boolean interrupted = false;
            //自旋
            for (;;) {
                final Node p = node.predecessor();
                if (p == head) {
                    int r = tryAcquireShared(arg);
                    if (r >= 0) {
                        //前驅結點為頭節點且成功獲取同步狀態 可退出自旋
                        setHeadAndPropagate(node, r);
                        p.next = null; // help GC
                        if (interrupted)
                            selfInterrupt();
                        failed = false;
                        return;
                    }
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }
private void setHeadAndPropagate(Node node, int propagate) {
        Node h = head; // Record old head for check below
        //退出自旋的節點變成首節點
        setHead(node);
        /*
         * Try to signal next queued node if:
         *   Propagation was indicated by caller,
         *     or was recorded (as h.waitStatus either before
         *     or after setHead) by a previous operation
         *     (note: this uses sign-check of waitStatus because
         *      PROPAGATE status may transition to SIGNAL.)
         * and
         *   The next node is waiting in shared mode,
         *     or we don't know, because it appears null
         *
         * The conservatism in both of these checks may cause
         * unnecessary wake-ups, but only when there are multiple
         * racing acquires/releases, so most need signals now or soon
         * anyway.
         */
        if (propagate > 0 || h == null || h.waitStatus < 0 ||
            (h = head) == null || h.waitStatus < 0) {
            Node s = node.next;
            if (s == null || s.isShared())
                doReleaseShared();
        }
    }

　　與獨占式一樣，共用式獲取也需要釋放同步狀態，通過調用releaseShared方法可以釋放同步狀態，代碼如下：

public final boolean releaseShared(int arg) {
        //釋放同步狀態
        if (tryReleaseShared(arg)) {
            //喚醒後續等待的節點
            doReleaseShared();
            return true;
        }
        return false;
    }
private void doReleaseShared() {
        /*
         * Ensure that a release propagates, even if there are other
         * in-progress acquires/releases.  This proceeds in the usual
         * way of trying to unparkSuccessor of head if it needs
         * signal. But if it does not, status is set to PROPAGATE to
         * ensure that upon release, propagation continues.
         * Additionally, we must loop in case a new node is added
         * while we are doing this. Also, unlike other uses of
         * unparkSuccessor, we need to know if CAS to reset status
         * fails, if so rechecking.
         */
        //自旋
　　　　for (;;) {
            Node h = head;
            if (h != null && h != tail) {
                int ws = h.waitStatus;
                if (ws == Node.SIGNAL) {
                    if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                        continue;            // loop to recheck cases
                    //喚醒後續節點
　　　　　　　　　　　　unparkSuccessor(h);
                }
                else if (ws == 0 &&
                         !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                    continue;                // loop on failed CAS
            }
            if (h == head)                   // loop if head changed
                break;
        }
    }

　　unparkSuccessor方法和獨占式是一樣的。

　　4、AQS應用

　　AQS被大量的應用在了同步工具上。

　　ReentrantLock：ReentrantLock類使用AQS同步狀態來保存鎖重覆持有的次數。當鎖被一個線程獲取時，ReentrantLock也會記錄下當前獲得鎖的線程標識，以便檢查是否是重覆獲取，以及當錯誤的線程試圖進行解鎖操作時檢測是否存在非法狀態異常。ReentrantLock也使用了AQS提供的ConditionObject，還向外暴露了其它監控和監測相關的方法。

　　ReentrantReadWriteLock：ReentrantReadWriteLock類使用AQS同步狀態中的16位來保存寫鎖持有的次數，剩下的16位用來保存讀鎖的持有次數。WriteLock的構建方式同ReentrantLock。ReadLock則通過使用acquireShared方法來支持同時允許多個讀線程。

　　Semaphore：Semaphore類（信號量）使用AQS同步狀態來保存信號量的當前計數。它裡面定義的acquireShared方法會減少計數，或當計數為非正值時阻塞線程；tryRelease方法會增加計數，在計數為正值時還要解除線程的阻塞。

　　CountDownLatch：CountDownLatch類使用AQS同步狀態來表示計數。當該計數為0時，所有的acquire操作（對應到CountDownLatch中就是await方法）才能通過。

　　FutureTask：FutureTask類使用AQS同步狀態來表示某個非同步計算任務的運行狀態（初始化、運行中、被取消和完成）。設置（FutureTask的set方法）或取消（FutureTask的cancel方法）一個FutureTask時會調用AQS的release操作，等待計算結果的線程的阻塞解除是通過AQS的acquire操作實現的。

　　SynchronousQueues：SynchronousQueues類使用了內部的等待節點，這些節點可以用於協調生產者和消費者。同時，它使用AQS同步狀態來控制當某個消費者消費當前一項時，允許一個生產者繼續生產，反之亦然。

       除了這些j.u.c提供的工具，還可以基於AQS自定義符合自己需求的同步器。

       AQS就學習到這，如果有描述不當的地方，還請留言交流。瞭解了AQS後下一步準備詳細學習基於AQS的工具類。

參考資料：

https://github.com/lingjiango/ConcurrentProgramPractice

http://gee.cs.oswego.edu/dl/papers/aqs.pdf

《Java併發編程的藝術》