如何理解AQS源碼

本篇內容介紹了“如何理解AQS源碼”的有關知識，在實際案例的操作過程中，不少人都會遇到這樣的困境，接下來就讓小編帶領大家學習一下如何處理這些情況吧！希望大家仔細閱讀，能夠學有所成！

AQS abstractQueueSynchronizer（抽象隊列同步器），是什么？

答：它是用來構建鎖 或者 其他同步器組件的重量級基礎框架，是整個JUC體系的基礎。通過內置FIFO隊列來完成獲取線程取鎖的排隊工作，并通過一個int類型變量標識持有鎖的狀態；

    前置知識點：

    1、可重入鎖（遞歸鎖）：

    sync（隱式鎖，jvm管理）和ReentrantLock（Lock顯式鎖，就是手動加解）是重入鎖的典型代表，為可以重復使用的鎖。一個變成多個流程，可以獲取同一把鎖。

    可重入鎖概念： 是指一個線程，在外層方法獲取鎖的時候，再次進入該線程的內層方法會自動獲取鎖（必須是同一個對象），不會被阻塞。可避免死鎖

    舉例： 遞歸調用同一個 sync修飾的方法或者代碼塊。必須是一個對象才行。一個線程調用一個對象的method1,method1 調用method2，method2調用method3, 3個方法都是被sync修飾，這樣也是一個可重入鎖的例子 。

    再比如下面這種

static Object lock = new Object();
public void mm(){
    synchronized (lock){
        System.out.println("===========mm method");
        synchronized (lock){
            System.out.println("=========== method");
        }
    }
}

     只有一個對象 和同步代碼塊，如果sycn中嵌套sync 并都是lock對象，那么該線程就會持有當前對象的鎖，并可重入。反編譯后發現

public void mm();
    Code:
       0: getstatic     #7                  // Field lock:Ljava/lang/Object;
       3: dup
       4: astore_1
       5: monitorenter
       6: getstatic     #8                  // Field java/lang/System.out:Ljava/io/PrintStream;
       9: ldc           #9                  // String ===========mm method
      11: invokevirtual #10                 // Method java/io/PrintStream.println:(Ljava/lang/String;)V
      14: aload_1
      15: monitorexit
      16: goto          24
      19: astore_2
      20: aload_1
      21: monitorexit
      22: aload_2
      23: athrow
      24: return
    Exception table:
       from    to  target type
           6    16    19   any
          19    22    19   any

    sync同步代碼塊加解鎖，使用的命令為monitorenter 和 monitorexit（同步方法標識是ACC_SYNCHRONIZED，在flag中）,enter 為加鎖，必須成對出現，但這里卻又兩個exit。原因為第一個exit為程序正常運行后的解鎖命令，并執行完后會執行goto到return ，也就是第24行，

    第二個exit 為當程序出現異常時，需要執行的解鎖命令；

如上就是可重入鎖的相關概念

    2、什么是LockSupport？

    根據jdk8 的api文檔顯示定義為： 用于創建鎖和其他同步類的基本線程阻塞原語； 

    是一個線程阻塞工具類，所有方法均為靜態，可以讓線程在任意位置阻塞，阻塞后也有對應的喚醒方法。

    先復習下object 對象的wait 和 notify 和Lock 的condition

• wait 和notify 必須在sync 代碼塊中才能使用，否則報錯。非法的監視器

• condition的await 和 signal方法也必須在lock 和unlock方法前執行，否則報錯，非法的監視器

• 線程一定要先 等待 ，再 被 喚醒，順序不能換

      LockSupport 有兩個關鍵函數 park 和unpark，該類使用了Permit（許可證）的概念來阻塞和喚醒線程的功能。每個線程都會有一個Permit，該Permit 只有兩個值 0 和1 ，默認是0。類似于信號量，但上限是1；

     來看park方法：

public static void park() {
    //unsafe的方法。初始為0
    UNSAFE.park(false, 0L);
}

    禁止當前線程進行線程調度，除非Permit可用，就是1

    如果Permit 為1（有可用證書） 將變更為0（線程仍然會處理業務邏輯），并且立即返回。否則當前線程對于線程調度目的將被禁用，并處于休眠狀態。直至發生三件事情之一：

• 一些其他線程調用當前線程作為目標的unpark ; 要么

• 其他一些線程當前線程為interrupts ; 要么

• 電話虛假（也就是說，沒有理由）返回。

     這種方法不報告是哪個線程導致該方法返回。 來電者應重新檢查導致線程首先停放的條件。 呼叫者還可以確定線程在返回時的中斷狀態。

     小結：Permit默認0，所以一開始調用park，當前線程被阻塞，直到別的線程將當前線程的Permit修改為1，從park方法處被喚醒，處理業務，然后會將permit修改為0，并返回；如果permit為1，調用park時會將permit修改為0，在執行業務邏輯到線程生命周期。與park方法定義吻合。

     在看unpark方法：

public static void unpark(Thread thread) {
    if (thread != null)
        UNSAFE.unpark(thread);
}

     在調用unpark方法后，會將Thread線程的許可permit設置成1，會自動喚醒thread線程，即，之前阻塞中的LockSupport.park方法會立即返回，然后線程執行業務邏輯 。 且 unpark可以在park之前執行。相當于執行park沒有效果。

    3、AQS abstractQueueSynchronizer 源碼

    剩余前置知識為： 公平鎖、非公平鎖、自旋鎖、鏈表、模板設計模式

     AQS使用volatile修飾的int類型的變量 標識鎖的狀態，通過內置的FIFO隊列來完成資源獲取的排隊工作，將每條要去搶占資源的線程封裝成node節點實現鎖的分配，通過CAS（自旋鎖）完成對state值的修改 ；

    （1）node節點源碼

static final class Node {
        /** Marker to indicate a node is waiting in shared mode */
    	//共享節點
        static final Node SHARED = new Node();
        /** Marker to indicate a node is waiting in exclusive mode */
    	//獨占節點
        static final Node EXCLUSIVE = null;

        /** waitStatus value to indicate thread has cancelled */
    	//線程被取消狀態
        static final int CANCELLED =  1;
        /** waitStatus value to indicate successor's thread needs unparking */
    	//	后續線程需要喚醒
        static final int SIGNAL    = -1;
        /** waitStatus value to indicate thread is waiting on condition */
    	//鄧丹condition喚醒
        static final int CONDITION = -2;
        /**
         * waitStatus value to indicate the next acquireShared should
         * unconditionally propagate
         */
    	//共享室同步狀態獲取 將會無條件傳播下去
        static final int PROPAGATE = -3;

        /**
         * Status field, taking on only the values:
         *   SIGNAL:     The successor of this node is (or will soon be)
         *               blocked (via park), so the current node must
         *               unpark its successor when it releases or
         *               cancels. To avoid races, acquire methods must
         *               first indicate they need a signal,
         *               then retry the atomic acquire, and then,
         *               on failure, block.
         *   CANCELLED:  This node is cancelled due to timeout or interrupt.
         *               Nodes never leave this state. In particular,
         *               a thread with cancelled node never again blocks.
         *   CONDITION:  This node is currently on a condition queue.
         *               It will not be used as a sync queue node
         *               until transferred, at which time the status
         *               will be set to 0. (Use of this value here has
         *               nothing to do with the other uses of the
         *               field, but simplifies mechanics.)
         *   PROPAGATE:  A releaseShared should be propagated to other
         *               nodes. This is set (for head node only) in
         *               doReleaseShared to ensure propagation
         *               continues, even if other operations have
         *               since intervened.
         *   0:          None of the above
         *
         * The values are arranged numerically to simplify use.
         * Non-negative values mean that a node doesn't need to
         * signal. So, most code doesn't need to check for particular
         * values, just for sign.
         *
         * The field is initialized to 0 for normal sync nodes, and
         * CONDITION for condition nodes.  It is modified using CAS
         * (or when possible, unconditional volatile writes).
         */
    	//初始為0，狀態是上面幾種，標識當前節點在隊列中的狀態
        volatile int waitStatus;

        /**
         * Link to predecessor node that current node/thread relies on
         * for checking waitStatus. Assigned during enqueuing, and nulled
         * out (for sake of GC) only upon dequeuing.  Also, upon
         * cancellation of a predecessor, we short-circuit while
         * finding a non-cancelled one, which will always exist
         * because the head node is never cancelled: A node becomes
         * head only as a result of successful acquire. A
         * cancelled thread never succeeds in acquiring, and a thread only
         * cancels itself, not any other node.
         */
    	//前置節點
        volatile Node prev;

        /**
         * Link to the successor node that the current node/thread
         * unparks upon release. Assigned during enqueuing, adjusted
         * when bypassing cancelled predecessors, and nulled out (for
         * sake of GC) when dequeued.  The enq operation does not
         * assign next field of a predecessor until after attachment,
         * so seeing a null next field does not necessarily mean that
         * node is at end of queue. However, if a next field appears
         * to be null, we can scan prev's from the tail to
         * double-check.  The next field of cancelled nodes is set to
         * point to the node itself instead of null, to make life
         * easier for isOnSyncQueue.
         */
    	//后置節點
        volatile Node next;

        /**
         * The thread that enqueued this node.  Initialized on
         * construction and nulled out after use.
         */
    	//當線程對象
        volatile Thread thread;

        /**
         * Link to next node waiting on condition, or the special
         * value SHARED.  Because condition queues are accessed only
         * when holding in exclusive mode, we just need a simple
         * linked queue to hold nodes while they are waiting on
         * conditions. They are then transferred to the queue to
         * re-acquire. And because conditions can only be exclusive,
         * we save a field by using special value to indicate shared
         * mode.
         */
        Node nextWaiter;

        /**
         * Returns true if node is waiting in shared mode.
         */
        final boolean isShared() {
            return nextWaiter == SHARED;
        }

        /**
         * Returns previous node, or throws NullPointerException if null.
         * Use when predecessor cannot be null.  The null check could
         * be elided, but is present to help the VM.
         *
         * @return the predecessor of this node
         */
        final Node predecessor() throws NullPointerException {
            Node p = prev;
            if (p == null)
                throw new NullPointerException();
            else
                return p;
        }

    node節點就是每一個等待執行的線程。還有一個waitState狀態字段，標識當前等待中的線程狀態

根據node節點 繪畫一個aqs基本結構圖

     解釋：state為狀態位，aqs為同步器。有head 和tail兩個 頭 尾節點，當state = 1時，表明同步器被占用（或者說當前有線程持有了同一個對象的鎖），將后續線程添加到隊列中，并用雙向鏈表連接，遵循FIFO。

     （2）以ReentrantLock的實現分析。因為他也實現了Lock 并內部持有同步器sync和AQS（以銀行柜臺例子）

     new ReentrantLock（）或 new ReentrantLock(false)時，創建的是非公平鎖,而 ReentrantLock對象內部還有 兩個類 分別為公平同步器和非公平同步器

static final class NonfairSync extends Sync
//公平鎖 有一個判斷隊列中是否有排隊的線程，這是與上面鎖不同的獲取方式
static final class FairSync extends Sync

    公平鎖解釋：先到先得，新線程在獲取鎖時，如果這個同步器的等待隊列中已經有線程在等待，那么當前線程會先進入等待隊列；

    非公平鎖解釋：新進來的線程不管是否有等待的線程，如果可以獲取鎖，則立刻占有鎖。

    這里還有一個關鍵的模板設計模式： 在查詢aqs的tryAcquire方法時發現，該方法直接拋出異常，這就是典型的模板設計模式，強制要求子類重寫該方法。否則不讓用

    1.1  當線程a到柜臺辦理業務時，會調用sync 的lock，即 a線程調用lock方法

final void lock() {
    //利用cas將當前對象的state 從0 設置成1，當然里面還有一個偏移量
    //意思就是如果是0 就設置為1成功返回true
    if (compareAndSetState(0, 1))
        setExclusiveOwnerThread(Thread.currentThread());
    else
        acquire(1);
}

    因為是第一個運行的線程，肯定是true所以，將當前運行的線程設置為a，即a線程占用了同步器，獲取了鎖

    1.2 當b線程運行lock時，發現不能將0設置成1（cas思想），就會運行acquire（1）方法

public final void acquire(int arg) {
    //這里用到了模板設計模式，強制子類實現該方法
    //因為默認使用非公平鎖，所以看NonfairSync 
    if (!tryAcquire(arg) &&
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
}

static final class NonfairSync extends Sync {
    private static final long serialVersionUID = 7316153563782823691L;

    /**
         * Performs lock.  Try immediate barge, backing up to normal
         * acquire on failure.
         */
    final void lock() {
        if (compareAndSetState(0, 1))
            setExclusiveOwnerThread(Thread.currentThread());
        else
            acquire(1);
    }

    protected final boolean tryAcquire(int acquires) {
        return nonfairTryAcquire(acquires);
    }
}

//該方法就是非公平鎖執行 等待的方法
final boolean nonfairTryAcquire(int acquires) {
    //獲取當前b線程
    final Thread current = Thread.currentThread();
    //獲取當前鎖的狀態是1，因為a線程已經獲取，并將state修改為1
    int c = getState();
    //有可能b在設置state時，a正辦理，到這兒時，a辦理完了。state為0了。
    if (c == 0) {
        //樂觀的將state 從0 修改為 1
        if (compareAndSetState(0, acquires)) {
            //設置當前獲取鎖的線程為b
            setExclusiveOwnerThread(current);
            return true;
        }
    }

    //有可能 a線程辦完業務。又回頭辦理了一個，所以當前線程持有鎖的線程依舊是a
    else if (current == getExclusiveOwnerThread()) {
        //2
        int nextc = c + acquires;
        if (nextc < 0) // overflow
            throw new Error("Maximum lock count exceeded");
        //設置state的值
        setState(nextc);
        return true;
    }
    //如果b線程走到這里，就證明b必須到等待隊列里去了
    return false;
}

     再來看邏輯運算符后面的邏輯

private Node addWaiter(Node mode) {
    //實例化一個節點，并將b 和 節點類型封裝成node
    Node node = new Node(Thread.currentThread(), mode);
    //等待隊列為null 所以tail初始化肯定是null
    //如果是線程c在b之后進來，tail就是b 節點
    Node pred = tail;
    //c節點之后都走這個方法
    if (pred != null) {
        //node的前置為tail
        //c 的前置設置為b
        node.prev = pred;
        //cas樂觀，比較 如果當前節點仍然是b 就將b 設置成c
        //b就是tail尾節點，將tail設置成c
        //這里根據源碼可知，就是將tail的值設置成c 并不影響pred的值，還是b
        if (compareAndSetTail(pred, node)) {
            //b 的下一個節點設置成c
            pred.next = node;
            return node;
        }
    }
    //線程b 入等待隊列
    enq(node);
    return node;
}

//入隊列方法
private Node enq(final Node node) {
    //該方法類似于while（true）
    for (;;) {
        //獲取tail節點
        Node t = tail;
        //初始化鎖等待隊列
        if (t == null) { // Must initialize
            //設置頭部節點為新的節點
            //這里看出，鎖等待隊列的第一個節點并非b，而是一個空node，該node為站位節點或者叫哨兵節點
            if (compareAndSetHead(new Node()))
                //將頭尾都指向該節點
                tail = head;
        } else {
            //第二次循環時，t為空node，將b的前置設置為空node
            node.prev = t;
            //設置tail節點為b節點
            if (compareAndSetTail(t, node)) {
                //空node節點的下一個節點為b node節點
                t.next = node;
                return t;
            }
        }
    }
}
/**
 * CAS head field. Used only by enq.
 */
private final boolean compareAndSetHead(Node update) {
    return unsafe.compareAndSwapObject(this, headOffset, null, update);
}

     以上b線程的進入等待隊列的操作就完成了 ，但線程還是活躍的，如何阻塞的呢？

     下面接著看acquireQueued方法

final boolean acquireQueued(final Node node, int arg) {
    boolean failed = true;
    try {
        boolean interrupted = false;
        //自旋執行
        for (;;) {
            //如果是b線程，這里p就是b節點的前置節點
            final Node p = node.predecessor();
            //空節點就是head節點，但又調用了一次tryAcquire方法，想再嘗試獲取鎖資源
            //如果a線程未處理完，那么這里返回false
            //如果a線程處理完成，那么這里就可以獲取到鎖
            if (p == head && tryAcquire(arg)) {
                //將head設置成b節點
                setHead(node);
                //原空節點的下連接斷開
                p.next = null; // help GC
                failed = false;
                return interrupted;
            }
            //第一次空節點進入should方法。返回false
            //當第二次循環到此處should方法返回true
            //執行parkAndCheckInterrupt方法，會將當前線程park，并獲取b線程的中斷狀態，如果未中斷返回false，并再次自旋一次 ，中斷為true
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    //head節點就是空節點所以w=0
    //空節點第二次進入時就是-1
    int ws = pred.waitStatus;
    if (ws == Node.SIGNAL)
        return true;
    //如果b節點狀態是其他，則將節點連接變化一下
    if (ws > 0) {
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;
    } else {
        //ws = 0時，使用cas將驗證pred 和ws 的值，是空節點和0  并將ws修改為-1
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}

private final boolean parkAndCheckInterrupt() {
    LockSupport.park(this);
    return Thread.interrupted();
}

     以上b線程未獲取鎖 并被掛起的操作就完成了

    1.3  當a線程調用unlock方法時：

public void unlock() {
    sync.release(1);
}

public final boolean release(int arg) {
    //判斷a線程是否完成業務。并釋放鎖，state=0
    if (tryRelease(arg)) {
        //獲取頭部節點，就是空節點
        Node h = head;
        //空節點在b獲取鎖時，狀態變更為-1，所以這里是true
        if (h != null && h.waitStatus != 0)
            //喚醒線程
            unparkSuccessor(h);
        return true;
    }
    return false;
}

//aqs父類的模板方法，強制要求子類實現該方法
protected boolean tryRelease(int arg) {
    throw new UnsupportedOperationException();
}

protected final boolean tryRelease(int releases) {
    //將鎖的狀態設置為0
    int c = getState() - releases;
    //判斷當前線程與 鎖的獨占線程是否一致
    if (Thread.currentThread() != getExclusiveOwnerThread())
        throw new IllegalMonitorStateException();
    //所得狀態標識
    boolean free = false;
    if (c == 0) {
        free = true;
        //如果state=0證明a完成了業務。那么鎖的獨占狀態就應該恢復為null
        setExclusiveOwnerThread(null);
    }
    //恢復鎖的state狀態
    setState(c);
    return free;
}

    注意：這里state是減1操作。如果ReentrantLock不斷可重入，那么這里是不能一次就歸零的。所以才會有ReentrantLock 調了幾次lock 就是要調幾次unlock

    a線程喚醒b線程

private void unparkSuccessor(Node node) {
    
    int ws = node.waitStatus;
    if (ws < 0)
        //空節點-1，設置成0
        compareAndSetWaitStatus(node, ws, 0);

	//獲取b節點，非null 且 waitState=0
    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)
        //將b線程喚醒
        LockSupport.unpark(s.thread);
}

     而 b線程還在acquireQueued方法里自旋呢，不過自旋后就會獲取鎖。

final boolean acquireQueued(final Node node, int arg) {
    boolean failed = true;
    try {
        boolean interrupted = false;
        //b線程在a未釋放鎖之前一直在自旋，
        for (;;) {
            final Node p = node.predecessor();
            //當a釋放鎖后，b獲取到鎖，將state設置為1
            //并將空節點的所有連接斷開等待GC回收
            //并返回當前線程 b 的中斷狀態
            if (p == head && tryAcquire(arg)) {
                setHead(node);
                p.next = null; // help GC
                failed = false;
                return interrupted;
            }
            if (shouldParkAfterFailedAcquire(p, node) &&
                //park當前線程b 并獲取b的中斷狀態，肯定是false，且調用的是帶參的native方法，多次調用會重置b線程的中斷狀態
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}
//調用該方法的if判斷如果進入了。會將b 中斷，并在不斷循環中再重置中斷狀態為false

     1.4 c線程的執行流程與b線程類似。會將b節點充等待隊列中移除，遵循FIFO

“如何理解AQS源碼”的內容就介紹到這里了，感謝大家的閱讀。如果想了解更多行業相關的知識可以關注億速云網站，小編將為大家輸出更多高質量的實用文章！