摘要：與之相關(guān)的方法有三個(gè)原子性地修改都是類型，可見我們可以進(jìn)行，來定義的獲取與釋放從而實(shí)現(xiàn)我們自定義的同步器。

源碼分析我認(rèn)為主要有兩個(gè)作用：滿足好奇心，我想每一個(gè)有追求的人都不會(huì)滿足于僅僅做一個(gè)API Caller實(shí)現(xiàn)功能就好，我們也想知道它到底是怎么實(shí)現(xiàn)的；借鑒與升華，當(dāng)我們明白了一個(gè)類的設(shè)計(jì)原理，在一定的情境下我們可以借鑒其設(shè)計(jì)哲學(xué)，甚至針對(duì)我們自己特殊的業(yè)務(wù)場(chǎng)景對(duì)其進(jìn)行改良與優(yōu)化。

下面我就以這篇文章開啟我的源碼閱讀之旅?？傮w而言，我會(huì)從這個(gè)類基本結(jié)構(gòu)入手，然后分析原理，再看看已有的應(yīng)用，并進(jìn)行分析與理解。

我之前一篇文章里提到過java的顯示鎖：ReentrantLock。此外，如果你編寫過并發(fā)程序那你一般也應(yīng)該用過CountDownLatch,Semaphore等等，這些都是同步器，而它們都基于AbstractQueuedSynchronizer（簡(jiǎn)稱AQS）實(shí)現(xiàn)的，那么我們今天就來看看這個(gè)牛逼的AQS是怎么實(shí)現(xiàn)這么多功能的。

首先打開IDEA，隨便新建一個(gè)類，然后輸入CountDownLatch，在它上面敲下Ctrl+B，就打開了CountDownLatch的源碼，然后發(fā)現(xiàn)有一個(gè)非常重要的靜態(tài)內(nèi)部類Sync繼承了AbstractQueuedSynchronizer，再次Ctrl+B，我們就打開了AQS的源碼，馬上就可以解開它的神秘面紗了，哼哼。

映入眼簾的首先就是大段大段的文檔，大意就是這個(gè)類 提供了一個(gè)基于FIFO隊(duì)列的實(shí)現(xiàn)了阻塞鎖和相關(guān)同步器（信號(hào)量，事件等）的框架...... 讀完了大概就了解這個(gè)類到底是怎么工作的了。下面我們開始分類型研究源碼，當(dāng)然不可能全部分析一遍，這里只把重點(diǎn)的列出來。
實(shí)際代碼分析中，我一般先看看這個(gè)結(jié)構(gòu)圖：

然后讀一讀開始的綜述文檔，然后從實(shí)例開始，像方法調(diào)用那樣依次深入查看，就能依次看到相關(guān)的方法、內(nèi)部類和屬性，還是Ctrl+B大法好啊，這屬于自底向上的源碼分析方法。如果直接從上面那張圖開始，對(duì)屬性、方法、內(nèi)部類挨個(gè)分析就屬于自頂向下的分析法了。我覺得對(duì)一個(gè)陌生的東西要想有清晰的認(rèn)知最好先自底向上捋一遍，便于搞清楚一個(gè)個(gè)具體功能的實(shí)現(xiàn)機(jī)制，然后再自頂向下看一遍，便于把控整體架構(gòu)，宏觀把握。這樣走兩遍再來總結(jié)一下就能比較透徹的掌握該技術(shù)了。

方法中，protected類型的一般要求具體的同步器子類來實(shí)現(xiàn)但是有些也可以直接用，public類型一般都是可以直接使用的當(dāng)然也可以自己實(shí)現(xiàn)，private就是AQS自己的內(nèi)部實(shí)現(xiàn)了，與具體子類無關(guān)。

一個(gè)private volatile int state;屬性代表了線程之間爭(zhēng)用的資源。與之相關(guān)的方法有三個(gè)

protected final int getState()
protected final void setState(int newState)
protected final boolean compareAndSetState(int expect, int update)//CAS原子性地修改state

都是protected類型，可見我們可以進(jìn)行Override，來定義state的獲取與釋放從而實(shí)現(xiàn)我們自定義的同步器。非常簡(jiǎn)單就不把全部源碼擺出來了。

這個(gè)queue是一個(gè)FIFO的隊(duì)列，每個(gè)節(jié)點(diǎn)都是下面的內(nèi)部類Node類型，等待著state這個(gè)資源，主要由兩個(gè)屬性決定private transient volatile Node head;和private transient volatile Node tail; 與之相關(guān)的方法有:

// 節(jié)點(diǎn)node進(jìn)入隊(duì)列，采用CAS的方式，返回其前驅(qū)
private Node enq(final Node node) {
    for (;;) {
        Node t = tail;
        if (t == null) { // 隊(duì)列為空，先初始化
            if (compareAndSetHead(new Node()))//設(shè)置頭結(jié)點(diǎn)
                tail = head;
        } else {// 隊(duì)列不為空
            node.prev = t;// 插入節(jié)點(diǎn)至隊(duì)列尾部
            if (compareAndSetTail(t, node)) {//CAS修改隊(duì)尾為node，之所以CAS是因?yàn)榭赡苡卸鄠€(gè)線程爭(zhēng)相入隊(duì)
                t.next = node;
                return t;
            }
        }
    }
}
// 將當(dāng)前線程以mode的方式（EXCLUSIVE或者SHARED）構(gòu)成新節(jié)點(diǎn)并入隊(duì)，返回這個(gè)新節(jié)點(diǎn)
private Node addWaiter(Node mode) {
    Node node = new Node(Thread.currentThread(), mode);
    // 更快的入隊(duì)方式，如果失敗再采用較慢的標(biāo)準(zhǔn)入隊(duì)方式enq
    Node pred = tail;
    if (pred != null) {
        node.prev = pred;
        if (compareAndSetTail(pred, node)) {
            pred.next = node;
            return node;
        }
    }
    enq(node);
    return node;
}
// 把node設(shè)置為新的頭，老的頭出隊(duì)
private void setHead(Node node) {
    head = node;
    node.thread = null;
    node.prev = null;
}

資源獲取分為EXCLUSIVE和SHARED兩種模式，對(duì)應(yīng)acquire與release、acquireShared與releaseShared。

首先是EXCLUSIVE資源獲取：

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

這里tryAcquire需要繼承類自己實(shí)現(xiàn)(成功true，失敗false)，如果tryAcquire成功則直接返回，否則addWaiter將當(dāng)前線程以獨(dú)占節(jié)點(diǎn)的方式置于同步隊(duì)列尾部等待。acquireQueued使得該節(jié)點(diǎn)等待獲取資源，一直獲取到資源才返回，整個(gè)等待過程中如果有中斷是不響應(yīng)的，但是獲取資源后會(huì)用selfInterrupt補(bǔ)上。

// 節(jié)點(diǎn)獲得資源才能返回否則一直自旋，中斷該線程不會(huì)實(shí)時(shí)響應(yīng)，但是如果被中斷過會(huì)返回true，否則返回false
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)) {// node前驅(qū)是頭結(jié)點(diǎn)，那么便可以嘗試去獲取資源了
                setHead(node);// 獲取成功，可以把node設(shè)為頭結(jié)點(diǎn)，也就是說頭結(jié)點(diǎn)是獨(dú)占資源的唯一擁有者
                p.next = null; // help GC
                failed = false;
                return interrupted;
            }
            // 走到這里說明獲取失敗，檢查是否應(yīng)該阻塞和中斷
            if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);//如果失敗了，就把waitStatus置為CANCELLED表示取消了
    }
}

// 獲取資源失敗后，當(dāng)前節(jié)點(diǎn)是否應(yīng)該阻塞
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    int ws = pred.waitStatus;
    if (ws == Node.SIGNAL)// 前驅(qū)pred獲得資源后會(huì)通知當(dāng)前節(jié)點(diǎn)node，所以可以放心的阻塞了（waitStatus會(huì)在下面內(nèi)部類解釋）
        return true;
    if (ws > 0) {// 前驅(qū)取消了資源獲取，那么當(dāng)前節(jié)點(diǎn)就要找到前面最近一個(gè)正在等待的節(jié)點(diǎn)
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;//此處 pred.waitStatus < 0，亦即pred 還在等待嘗試獲取資源
    } else {// 前驅(qū)正在等待，則設(shè)置其狀態(tài)為SIGNAL，讓他獲取資源后通知本節(jié)點(diǎn)，
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        // 但是本節(jié)點(diǎn)不能馬上阻塞，因?yàn)樵O(shè)置不一定能成功,需要下次再次檢查
    }
    return false;
}

// 阻塞本線程。被喚醒后要返回本線程是否被中斷過。
private final boolean parkAndCheckInterrupt() {
    LockSupport.park(this);
    return Thread.interrupted();
}

    

然后是EXCLUSIVE資源釋放：

public final boolean release(int arg) {
    if (tryRelease(arg)) {
        Node h = head;
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        return true;
    }
    return false;
}

與上面相對(duì)應(yīng)，這里tryRelease也需要繼承類自己實(shí)現(xiàn)(成功true，失敗false)，如果釋放成功，則調(diào)用unparkSuccessor喚醒后繼節(jié)點(diǎn)返回true，否則返回false。

private void unparkSuccessor(Node node) {
    int ws = node.waitStatus;
    if (ws < 0)// 可能需要釋放通知信號(hào)，把狀態(tài)置零，允許失敗
        compareAndSetWaitStatus(node, ws, 0);

    Node s = node.next;// 找到后繼節(jié)點(diǎn)
    if (s == null || s.waitStatus > 0) {// 如果后繼節(jié)點(diǎn)為空或者已經(jīng)取消
        s = null;// 確保該節(jié)點(diǎn)的釋放
        for (Node t = tail; t != null && t != node; t = t.prev)// 從隊(duì)尾開始找到需要通知的最近的后繼節(jié)點(diǎn)
            if (t.waitStatus <= 0)
                s = t;
    }
    if (s != null)// 如果需喚醒的后繼節(jié)點(diǎn)存在則喚醒之
        LockSupport.unpark(s.thread);
}

再看SHARED資源獲取：

public final void acquireShared(int arg) {
    if (tryAcquireShared(arg) < 0)
        doAcquireShared(arg);
}

這里tryAcquireShared也需要自己實(shí)現(xiàn)（負(fù)值說明失敗，非負(fù)值表示獲取成功后剩下的可用資源數(shù)），如果獲取失敗就調(diào)用doAcquireShared進(jìn)入同步隊(duì)列等待。

// 等待獲取共享資源時(shí)不響應(yīng)中斷，但是獲取資源成功后會(huì)用selfInterrupt補(bǔ)上
private void doAcquireShared(int arg) {
    final Node node = addWaiter(Node.SHARED);// 入隊(duì)尾
    boolean failed = true;
    try {
        boolean interrupted = false;
        for (;;) {
            final Node p = node.predecessor();
            if (p == head) {// 處于隊(duì)列第二個(gè)位置，可以嘗試獲取資源
                int r = tryAcquireShared(arg);
                if (r >= 0) {// 獲取成功
                    setHeadAndPropagate(node, r);// 將自己設(shè)為隊(duì)列頭，并喚醒可能獲取資源的后面幾個(gè)節(jié)點(diǎn)
                    p.next = null; // help GC
                    if (interrupted)
                        selfInterrupt();
                    failed = false;
                    return;
                }
            }
            if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt())// 同acquireQueued的分析
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

private void setHeadAndPropagate(Node node, int propagate) {
    Node h = head; // 舊的頭
    setHead(node); // 設(shè)置新的頭
    // 如果還有資源，則喚醒下一個(gè)，采用保守策略，多喚醒幾次即使沒獲取到資源也無所謂，盡量做到不漏掉資源
    if (propagate > 0 || h == null || h.waitStatus < 0 || (h = head) == null || h.waitStatus < 0) {        
        Node s = node.next;
        if (s == null || s.isShared())
            doReleaseShared();
    }
}

最后SHARED資源釋放：

public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) {
        doReleaseShared();
        return true;
    }
    return false;
}
   

這里tryReleaseShared依然要自己實(shí)現(xiàn)（如果可以允許下一個(gè)節(jié)點(diǎn)獲得資源則返回true，否則false），如果釋放成功則調(diào)用doReleaseShared喚醒后繼節(jié)點(diǎn)。需要注意的是tryReleaseShared由于可能多個(gè)線程并發(fā)操作所以一般需要CAS而tryRelease不需要。

private void doReleaseShared() {
    for (;;) {
        Node h = head;
        if (h != null && h != tail) {
            int ws = h.waitStatus;
            if (ws == Node.SIGNAL) {// 需要喚醒
                if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))//設(shè)置WaitStatus失敗
                    continue;            
                unparkSuccessor(h);// 一定要設(shè)置成功才喚醒
            }
            else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                continue;// CAS設(shè)置失敗則繼續(xù)循環(huán)
        }
        if (h == head)// 頭變了，不需要繼續(xù)喚醒
            break;
    }
}

此外，資源獲取除了一直等待的方式之外還有對(duì)應(yīng)的限制等待時(shí)間的方法如tryAcquire與tryAcquireNanos，不必多言，釋放就只有一直等而沒有限制等待時(shí)間的了。也有響應(yīng)中斷與不響應(yīng)的對(duì)應(yīng)，如acquireInterruptibly與acquire，差別不大，不必多言。

等待隊(duì)列的節(jié)點(diǎn)類，等待隊(duì)列是CLH（Craig,Landin,Hagersten）鎖隊(duì)列的一種變體，CLH鎖通常用來作為自旋鎖。

每個(gè)節(jié)點(diǎn)主要維護(hù)了下面一些狀態(tài)

對(duì)應(yīng)的線程thread

等待狀態(tài)waitStatus 含，0：初始狀態(tài)；CANCELLED 1：被取消；SIGNAL -1：當(dāng)前線程釋放資源或取消后需要喚醒后繼節(jié)點(diǎn)；CONDITION -2：條件等待；PROPAGATE -3：下一個(gè)acquireShared操作應(yīng)該被無條件傳播。實(shí)際使用中，一般只關(guān)注正負(fù)，非負(fù)數(shù)就意味著節(jié)點(diǎn)不需要釋放信號(hào)

資源獲取模式有SHARED（默認(rèn)）和EXCLUSIVE兩個(gè)

同步隊(duì)列中的前驅(qū)后繼節(jié)點(diǎn)prev和next

作為同步隊(duì)列節(jié)點(diǎn)時(shí)，nextWaiter有：EXCLUSIVE、SHARED標(biāo)識(shí)當(dāng)前節(jié)點(diǎn)是獨(dú)占模式還是共享模式；與ConditionObject搭配使用作為條件等待隊(duì)列節(jié)點(diǎn)時(shí)，nextWaiter保存后繼節(jié)點(diǎn)。所以實(shí)際上這個(gè)Node類是被復(fù)用了，既用于同步隊(duì)列，也用于條件等待隊(duì)列。

這個(gè)類實(shí)現(xiàn)了Condition接口，主要用來完成常見的條件等待、喚醒等操作。一個(gè)ConditionObject 包含一個(gè)等待隊(duì)列，由firstWaiter和lastWaiter決定。當(dāng)前線程調(diào)用Condition.await()方法時(shí)，會(huì)被構(gòu)造成為節(jié)點(diǎn)，然后置于條件等待隊(duì)列隊(duì)尾。
我們看最常用的條件等待方法

    
    // 條件等待，響應(yīng)中斷
    public final void await() throws InterruptedException {
        if (Thread.interrupted())
            throw new InterruptedException();// 響應(yīng)中斷
        Node node = addConditionWaiter();// 當(dāng)前線程加入條件等待隊(duì)列
        int savedState = fullyRelease(node);// 釋放資源，并獲得本節(jié)點(diǎn)需要的資源數(shù)，以便再次獲取
        int interruptMode = 0;
        while (!isOnSyncQueue(node)) {// 如果當(dāng)前節(jié)點(diǎn)不是在同步隊(duì)列，也就是還在等待隊(duì)列等待著條件發(fā)生
            LockSupport.park(this);// 就會(huì)一直阻塞
            if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                break;
        }
        if (acquireQueued(node, savedState) && interruptMode != THROW_IE)//加入同步隊(duì)列等待獲取資源，成功后要檢查中斷響應(yīng)
            interruptMode = REINTERRUPT;
        if (node.nextWaiter != null) // clean up if cancelled
            unlinkCancelledWaiters();
        if (interruptMode != 0)
            reportInterruptAfterWait(interruptMode);
    }

    private Node addConditionWaiter() {
        Node t = lastWaiter;
        // 如果最后一個(gè)條件等待節(jié)點(diǎn)是取消的狀態(tài)
        if (t != null && t.waitStatus != Node.CONDITION) {
            unlinkCancelledWaiters();// 清理整個(gè)鏈路的無效節(jié)點(diǎn)
            t = lastWaiter;
        }
        //以條件等待的方式將當(dāng)前線程封裝成節(jié)點(diǎn)
        Node node = new Node(Thread.currentThread(), Node.CONDITION);
        if (t == null)//條件等待隊(duì)列為空就初始化
            firstWaiter = node;
        else// 隊(duì)列不空，插入隊(duì)尾
            t.nextWaiter = node;
        lastWaiter = node;
        return node;// 返回新插入的節(jié)點(diǎn)
    }

    final int fullyRelease(Node node) {
        boolean failed = true;
        try {
            int savedState = getState();// 本節(jié)點(diǎn)需要的資源數(shù)
            if (release(savedState)) {// 釋放掉這么多資源
                failed = false;
                return savedState;
            } else {
                throw new IllegalMonitorStateException();
            }
        } finally {
            if (failed)
                node.waitStatus = Node.CANCELLED;
        }
    }

然后是信號(hào)方法：

    
    public final void signal() {
        if (!isHeldExclusively())// 要使用該方法必須先是獨(dú)占線程
            throw new IllegalMonitorStateException();
        Node first = firstWaiter;
        if (first != null)
            doSignal(first);//找到第一個(gè)條件等待節(jié)點(diǎn)，并發(fā)出信號(hào)
    }
    
    // 去掉條件等待隊(duì)列的節(jié)點(diǎn)，直到遇上沒取消的或者空節(jié)點(diǎn)
    private void doSignal(Node first) {
        do {
            if ( (firstWaiter = first.nextWaiter) == null)
                lastWaiter = null;
            first.nextWaiter = null;
        } while (!transferForSignal(first) &&
                 (first = firstWaiter) != null);
    }
    
    final boolean transferForSignal(Node node) {
        if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))// 節(jié)點(diǎn)已被取消
            return false;
        Node p = enq(node);// 條件等待隊(duì)列的第一個(gè)節(jié)點(diǎn)被加入同步隊(duì)列的隊(duì)尾
        int ws = p.waitStatus;
        if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
            LockSupport.unpark(node.thread);// 喚醒節(jié)點(diǎn)對(duì)應(yīng)線程
        return true;
    }

下面用兩個(gè)例子來看看AQS的具體使用場(chǎng)景，分別是使用獨(dú)占模式的ReentrantLock和共享模式的CountDownLatch 。
一般使用AQS的類，都會(huì)用一個(gè)內(nèi)部類Sync來繼承AQS，并實(shí)現(xiàn)那幾個(gè)protected的方法。

ReentrantLock 有FairSync和NonfairSync兩個(gè)類來實(shí)現(xiàn)公平鎖和非公平鎖，我們看非公平鎖，主要幾個(gè)方法是

lock()，使得NonfairSync調(diào)用compareAndSetState把state從0設(shè)為1并用setExclusiveOwnerThread把當(dāng)前線程設(shè)為獨(dú)占線程（亦即首次獲得鎖），如果失敗則使用acquire(1)調(diào)用nonfairTryAcquire?？傮w流程就是如果state為0，那么就是本線程首次獲得鎖，把state置為1，否則如果當(dāng)前線程是獨(dú)占線程則將state+1（這也是鎖可重入的關(guān)鍵），如果都不是就進(jìn)入acquireQueued流程等待獲得鎖了了

unlock()，調(diào)用AQS的release(1)方法，實(shí)際上是調(diào)用了Sync的tryRelease(1)方法，如果state-1為0，那么返回true，否則返回false。也就是說，重入鎖必須釋放夠重入次數(shù)才算真正釋放成功，但是unlock()方法本身不會(huì)管這個(gè)最終結(jié)果，只管釋放

tryLock()，與lock()區(qū)別是不等待，立即返回，只有喚醒時(shí)就是獨(dú)占線程才能返回true，實(shí)現(xiàn)方法是nonfairTryAcquire

newCondition()直接返回了了AQS的內(nèi)部類ConditionObject

isLocked() 如果state為0則表示未加鎖返回false，否則返回true

CountDownLatch 主要幾個(gè)方法是

CountDownLatch(int count)，構(gòu)造方法，設(shè)置 AQS 的 state 為 count

await()，調(diào)用 AQS 的 acquireSharedInterruptibly(int arg) 方法，然后調(diào)用自己覆蓋的tryAcquireShared(int acquires)來獲得state的值是否為0，如果是0就結(jié)束等待直接返回了，如果不是0就調(diào)用 AQS 的 doAcquireSharedInterruptibly(int arg)方法，該方法會(huì)循環(huán)等待，直到state為0才返回或者被中斷。

countDown()，調(diào)用 AQS 的 releaseShared(int arg) 方法，實(shí)際上是調(diào)用了自己覆蓋的 tryReleaseShared(int releases) 方法，把 state 減了1，如果此時(shí)state為0，則調(diào)用 AQS 的doReleaseShared()方法

總體而言，AQS提供了一個(gè)模板方法模式，將獲得鎖釋放鎖一些必要的流程操作都規(guī)定好了，我們只需要填充一些具體的獲得與釋放方法

getState(),setState(int newState),compareAndSetState(int expect,int update)：是資源相關(guān)操作，保證原子性

tryAcquire(int arg)：嘗試獨(dú)占獲取資源。成功返回true，失敗返回false。

tryRelease(int arg)：嘗試獨(dú)占釋放資源。成功返回true，失敗返回false。

tryAcquireShared(int arg)：嘗試共享獲取資源。負(fù)數(shù)表示失敗，非負(fù)數(shù)表示成功代表剩余可用資源

tryReleaseShared(int arg)：嘗試共享釋放資源。如果釋放后可以喚醒后續(xù)等待結(jié)點(diǎn)返回true，否則返回false。

isHeldExclusively()：代表當(dāng)前線程是否獨(dú)占資源，只有用到Condition之時(shí)才需要去實(shí)現(xiàn)它。

自定義同步器時(shí)，一般都是自己寫一個(gè) static class Sync extends AbstractQueuedSynchronizer 靜態(tài)內(nèi)部類來實(shí)現(xiàn)具體的方法。

閱讀原文：MageekChiu