Java並發之AQS原理淺析上

        鎖是最常用的同步方法之一，在高並發的環境下激烈的鎖競爭會導致程序的性能下降，所以我們自然有必要深入的學習一下鎖的相關知識。

         java的內置鎖一直都是備受爭議的，在JDK 1.6之前，synchronized這個重量級鎖其性能一直都是較為低下，雖然在1.6后，進行大量的鎖優化策略，如自適應自旋，鎖消除，鎖粗化，輕量級鎖，偏向鎖等等,但是與Lock相比synchronized還是存在一些缺陷的：雖然synchronized提供了便捷性的隱式獲取鎖釋放鎖機制（基於JVM機制），但是它卻缺少了獲取鎖與釋放鎖的可操作性，可中斷、超時獲取鎖，且它為獨占式在高並發場景下性能大打折扣。

         在《深入理解Java虛擬機》這本書上,作者說了這句話:與其說ReentrantLock性能好,還不如說synchronized還有很大優化的余地。在JDK1.6之后,人們發現synchronized與ReentrantLock的性能基本上是完全持平的(但是在JDK是1.8做基礎測試時synchronized的性能還是不如ReentrantLock,原因暫未發現)。虛擬機在未來的性能改進中肯定會更加偏向於原生的synchronized,所以還是提倡synchronized能實現需求的情況下,優先考慮使用synchronized來進行同步。但是在jdk1.8測試中lock的使用比當確實使用synchroinzed同步時我們的性能瓶頸時,我們可以用ReentrantLock來進行性能的測試,如果確實更優,我們就可以選擇用ReetrantLock來進行同步。

        在介紹Lock之前，我們需要先熟悉一個非常重要的基礎組件，JUC包下的核心基礎組件。也是實現大部分同步需求的基礎。學習該組件是學習JUC繞不開的一塊內容。該組件就是AQS。

AQS簡介

• AQS：AbstractQueuedSynchronizer，即隊列同步器。它是構建鎖或者其他同步組件的基礎框架（如ReentrantLock、ReentrantReadWriteLock、Semaphore等）。
• AQS解決了子類實現同步器時涉及當的大量細節問題，例如獲取同步狀態、FIFO同步隊列。基於AQS來構建同步器可以帶來很多好處。自定義同步器在實現時只需要實現共享資源state的獲取與釋放方式即可，至於具體線程等待隊列的維護（如獲取資源失敗入隊/喚醒出隊等），AQS已經在頂層實現好了，所以使用AQS不僅能夠極大地減少實現工作，而且也不必處理在多個位置上發生的競爭問題。
• 在基於AQS構建的同步器中，只能在一個時刻發生阻塞，從而降低上下文切換的開銷，提高了吞吐量。同時在設計AQS時充分考慮了可伸縮行，因此J.U.C中所有基於AQS構建的同步器均可以獲得這個優勢。
• AQS的主要使用方式是繼承，子類通過繼承同步器並實現它的抽象方法來管理同步狀態。
• AQS使用一個int類型的成員變量state來表示同步狀態，當state>0時表示已經獲取了鎖，當state = 0時表示釋放了鎖。它提供了三個方法（getState()、setState(int newState)、compareAndSetState(int expect,int update)）來對同步狀態state進行操作，當然AQS可以確保對state的操作是安全的。
• AQS通過內置的FIFO同步隊列來完成資源獲取線程的排隊工作，如果當前線程獲取同步狀態失敗（鎖）時，AQS則會將當前線程以及等待狀態等信息構造成一個節點（Node）並將其加入同步隊列，同時會阻塞當前線程，當同步狀態釋放時，則會把節點中的線程喚醒，使其再次嘗試獲取同步狀態。

AQS常用方法

    關於state的方法主要有一下三種

• getState()：返回同步狀態的當前值；

• setState(int newState)：設置當前同步狀態；

• compareAndSetState(int expect, int update)：使用CAS設置當前狀態，該方法能夠保證狀態設置的原子性；

   自定義同步器實現時主要實現以下幾種方法

• tryAcquire(int arg)：獨占式獲取同步狀態，獲取同步狀態成功后，其他線程需要等待該線程釋放同步狀態才能獲取同步狀態

• tryRelease(int arg)：獨占式釋放同步狀態；

• tryAcquireShared(int arg)：共享式獲取同步狀態，返回值大於等於0則表示獲取成功，否則獲取失敗；

• tryReleaseShared(int arg)：共享式釋放同步狀態；

• isHeldExclusively()：當前同步器是否在獨占式模式下被線程占用，一般該方法表示是否被當前線程所獨占；

其余方法

• acquire(int arg)：獨占式獲取同步狀態，如果當前線程獲取同步狀態成功，則由該方法返回，否則，將會進入同步隊列等待，該方法將會調用可重寫的tryAcquire(int arg)方法；

• acquireInterruptibly(int arg)：與acquire(int arg)相同，但是該方法響應中斷，當前線程為獲取到同步狀態而進入到同步隊列中，如果當前線程被中斷，則該方法會拋出InterruptedException異常並返回；

• tryAcquireNanos(int arg,long nanos)：超時獲取同步狀態，如果當前線程在nanos時間內沒有獲取到同步狀態，那么將會返回false，已經獲取則返回true；

• acquireShared(int arg)：共享式獲取同步狀態，如果當前線程未獲取到同步狀態，將會進入同步隊列等待，與獨占式的主要區別是在同一時刻可以有多個線程獲取到同步狀態；

• acquireSharedInterruptibly(int arg)：共享式獲取同步狀態，響應中斷；

• tryAcquireSharedNanos(int arg, long nanosTimeout)：共享式獲取同步狀態，增加超時限制；

• release(int arg)：獨占式釋放同步狀態，該方法會在釋放同步狀態之后，將同步隊列中第一個節點包含的線程喚醒；

• releaseShared(int arg)：共享式釋放同步狀態；

CLH

       CLH同步隊列是一個FIFO雙向隊列，AQS依賴它來完成同步狀態的管理，當前線程如果獲取同步狀態失敗時，AQS則會將當前線程已經等待狀態等信息構造成一個節點（Node）並將其加入到CLH同步隊列，同時會阻塞當前線程，當同步狀態釋放時，會把首節點喚醒（公平鎖），使其再次嘗試獲取同步狀態。

在CLH同步隊列中，一個節點表示一個線程，它保存着線程的引用（thread）、狀態（waitStatus）、前驅節點（prev）、后繼節點（next），其數據結構如下

                                                       

 

   其實就是個雙端雙向鏈表。

   數據定義如下

static final class Node {
    /** 共享 */
    static final Node SHARED = new Node();
    /** 獨占 */
    static final Node EXCLUSIVE = null;
    /**
     * 因為超時或者中斷，節點會被設置為取消狀態，被取消的節點時不會參與到競爭中的，他會一直保持取消狀態不會轉變為其他狀態；
     */
    static final int CANCELLED =  1;
    /**
     * 后繼節點的線程處於等待狀態，而當前節點的線程如果釋放了同步狀態或者被取消，將會通知后繼節點，使后繼節點的線程得以運行
     */
    static final int SIGNAL    = -1;
    /**
     * 節點在等待隊列中，節點線程等待在Condition上，當其他線程對Condition調用了signal()后，改節點將會從等待隊列中轉移到同步隊列中，加入到同步狀態的獲取中
     */
    static final int CONDITION = -2;
    /**
     * 表示下一次共享式同步狀態獲取將會無條件地傳播下去
     */
    static final int PROPAGATE = -3;
    /** 等待狀態 */
    volatile int waitStatus;
    /** 前驅節點 */
    volatile Node prev;
    /** 后繼節點 */
    volatile Node next;
    /** 獲取同步狀態的線程 */
    volatile Thread thread;
    Node nextWaiter;
    final boolean isShared() {
        return nextWaiter == SHARED;
    }
    final Node predecessor() throws NullPointerException {
        Node p = prev;
        if (p == null)
            throw new NullPointerException();
        else
            return p;
    }
    Node() {
    }
    Node(Thread thread, Node mode) {
        this.nextWaiter = mode;
        this.thread = thread;
    }
    Node(Thread thread, int waitStatus) {
        this.waitStatus = waitStatus;
        this.thread = thread;
    }
}

    可以看到AQS支持兩種同步模式，分別是Exclusive（獨占，只有一個線程能執行，如ReentrantLock）和Share（共享，多個線程可同時執行，如Semaphore/CountDownLatch）。這樣方便使用者實現不同類型的同步組件。簡而言之，AQS為使用者提供了多樣的底層支撐，具體如何組裝實現，使用者可以自由發揮。

   入列

   CHL這種鏈表式結構入列，無非就是tail指向新節點、新節點的前驅節點指向當前最后的節點，當前最后一個節點的next指向當前節點，直接看源碼相關操作在addWaiter(Node node)方法里。此方法用於將當前線程加入到等待隊列的隊尾，並返回當前線程所在的結點  

    private Node addWaiter(Node mode) {
        //根據給定的模式（獨占或者共享）新建Node
        Node node = new Node(Thread.currentThread(), mode);
        //快速嘗試添加尾節點
        Node pred = tail;
        if (pred != null) {
            node.prev = pred;
            //CAS設置尾節點
            if (compareAndSetTail(pred, node)) {
                pred.next = node;
                return node;
            }
        }
        //多次嘗試
        enq(node);
        return node;
    }

    addWaiter(Node node)先通過快速嘗試設置尾節點，如果失敗，則調用enq(Node node)方法設置尾節點

  private Node enq(final Node node) {
        //多次嘗試，直到成功為止
        for (;;) {
            Node t = tail;
            //tail不存在，設置為首節點
            if (t == null) {
                if (compareAndSetHead(new Node()))
                    tail = head;
            } else {
                //設置為尾節點
                node.prev = t;
                if (compareAndSetTail(t, node)) {
                    t.next = node;
                    return t;
                }
            }
        }
    }

        此方法用於將node加入隊尾，該方法核心就是通過CAS自旋的方式來設置尾節點，知道獲得預期的結果即添加節點成功，當前線程才會返回。（這種方式很經典AtomicInteger.getAndIncrement()方法也是這樣做的）

     出列

     CLH同步隊列遵循FIFO（先進先出），首節點的線程釋放同步狀態后，將會喚醒它的后繼節點（next），而后繼節點將會在獲取同步狀態成功時將自己設置為首節點，這個過程非常簡單，head執行該節點並斷開原首節點的next和當前節點的prev即可，注意在這個過程是不需要使用CAS來保證的，因為只有一個線程能夠成功獲取到同步狀態。

同步狀態的獲取與釋放

       AQS的設計模式采用的模板方法模式，子類通過繼承的方式，實現它的抽象方法來管理同步狀態，對於子類而言它並沒有太多的活要做，AQS提供了大量的模板方法來實現同步，主要是分為三類：獨占式獲取和釋放同步狀態、共享式獲取和釋放同步狀態、查詢同步隊列中的等待線程情況。自定義子類使用AQS提供的模板方法就可以實現自己的同步語義。

    獨占式同步狀態獲取

    此方法是獨占模式下線程獲取共享資源的頂層入口。如果獲取到資源，線程直接返回，否則進入等待隊列，直到獲取到資源為止，且整個過程忽略中斷的影響。這也正是lock()的語義，當然不僅僅只限於lock()。也就是說由於線程獲取同步狀態失敗加入到CLH同步隊列中，后續對線程進行中斷操作時，線程不會從同步隊列中移除獲取到資源后。下面是acquire()的源碼：

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

• tryAcquire：去嘗試獲取鎖，獲取成功則設置鎖狀態並返回true，否則返回false。該方法由自定義同步組件自己實現（通過state的get/set/CAS），該方法必須要保證線程安全的獲取同步狀態。

• addWaiter：如果tryAcquire返回FALSE（獲取同步狀態失敗），則調用該方法將當前線程加入到CLH同步隊列尾部，並標記為獨占模式。

• acquireQueued：當前線程會根據公平性原則來進行阻塞等待（自旋）,直到獲取鎖為止；如果在整個等待過程中被中斷過，則返回true，否則返回false。

• selfInterrupt：如果線程在等待過程中被中斷過，它是不響應的。只是獲取資源后才再進行自我中斷selfInterrupt()，將中斷補上。

 tryAcquire(int)

protected boolean tryAcquire(int arg) {
        throw new UnsupportedOperationException();
    }

      該方法直接拋出異常，具體實現交自定義同步器類實現。這里之所以沒有定義成abstract，是因為獨占模式下只用實現tryAcquire-tryRelease，而共享模式下只用實現tryAcquireShared-tryReleaseShared。如果都定義成abstract，那么每個模式也要去實現另一模式下的接口。

 acquireQueued

   在執行到此方法時已經說明一點：該線程獲取資源失敗，已經被放入等待隊列尾部了。所以 acquireQueued方法就是讓線程進入等待狀態休息，直到其他線程徹底釋放資源后喚醒該線程，獲取所需資源，然后執行該線程所需執行的任務。

   acquireQueued方法為一個自旋的過程，也就是說當前線程（Node）進入同步隊列后，就會進入一個自旋的過程，每個節點都會自我觀察，當條件滿足，獲取到同步狀態后，就可以從這個自旋過程中退出，否則會一直執行下去。

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)) {
                    /*將head指向該節點*/
                    setHead(node);
                   /* 方便GC回收垃圾 */
                    p.next = null; 
                    failed = false;
                   /*返回等待過程中是否被中斷過*/
                    return interrupted;
                }
                /*獲取失敗，線程就進入waiting狀態，直到被unpark()*/
                if (shouldParkAfterFailedAcquire(p, node) &&parkAndCheckInterrupt())
                    /*如果等待過程中被中斷過一次，就標記為true*/
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

       從上面代碼中可以看到，當前線程會一直嘗試獲取同步狀態，當然前提是只有其前驅節點為頭結點才能夠嘗試獲取同步狀態，理由：

• 保持FIFO同步隊列原則。

• 頭節點釋放同步狀態后，將會喚醒其后繼節點，后繼節點被喚醒后需要檢查自己是否為頭節點。

shouldParkAfterFailedAcquire(Node, Node)

    此方法主要用於檢查狀態，查看當前節點是否進入waiting狀態

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    int ws = pred.waitStatus;//拿到前驅節點的狀態
    if (ws == Node.SIGNAL)
        //狀態為SIGNAL,如果前驅節點處於等待狀態，直接返回true
        return true;
    if (ws > 0) {
        /*
         * 如果前驅節點放棄了，那就一直往前找，直到找到最近一個正常等待的狀態，並排在它的后邊。
         * 注意：那些放棄的結點，由於被自己“加塞”到它們前邊，它們相當於形成一個無引用鏈，稍后就會被GC回收
         */
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;
    } else {
         //如果前驅節點正常，那就把前驅的狀態通過CAS的方式設置成SIGNAL
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}

 這段代碼主要檢查當前線程是否需要被阻塞，具體規則如下：

1. 如果當前線程的前驅節點狀態為SINNAL，則表明當前線程需要被阻塞，調用unpark()方法喚醒，直接返回true，當前線程阻塞

2. 如果當前線程的前驅節點狀態為CANCELLED（ws > 0），則表明該線程的前驅節點已經等待超時或者被中斷了，則需要從CLH隊列中將該前驅節點刪除掉，直到回溯到前驅節點狀態 <= 0 ，返回false

3. 如果前驅節點非SINNAL，非CANCELLED，則通過CAS的方式將其前驅節點設置為SINNAL，返回false

      整個流程中，如果前驅結點的狀態不是SIGNAL，那么自己就不能被阻塞，需要去找個安心的休息點（前驅節點狀態 <= 0 ），同時可以再嘗試下看有沒有機會去獲取資源。

     如果 shouldParkAfterFailedAcquire(Node pred, Node node) 方法返回true，則調用parkAndCheckInterrupt()方法阻塞當前線程： 

private final boolean parkAndCheckInterrupt() {
        //調用park()使線程進入waiting狀態
          LockSupport.park(this); 
          //如果被喚醒，查看自己是不是被中斷的
          return Thread.interrupted();
 }

    parkAndCheckInterrupt() 方法主要是把當前線程掛起，從而阻塞住線程的調用棧，同時返回當前線程的中斷狀態。

實現細節有點復雜 未完待續