摘要：獲取鎖的過程當(dāng)線程調(diào)用申請獲取鎖資源，如果成功，則進(jìn)入臨界區(qū)。如果隊列中有其他等待鎖資源的線程需要喚醒，則喚醒隊列中的第一個等待節(jié)點(diǎn)先入先出。釋放鎖時，如果隊列中有等待的線程就進(jìn)行喚醒。

每一個Java工程師應(yīng)該都或多或少了解過AQS，我自己也是前前后后，反反復(fù)復(fù)研究了很久，看了忘，忘了再看，每次都有不一樣的體會。這次趁著寫博客，打算重新拿出來系統(tǒng)的研究下它的源碼，總結(jié)成文章，便于以后復(fù)習(xí)。

原文地址：http://www.jianshu.com/p/7144...

AbstractQueuedSynchronizer（以下簡稱AQS）作為java.util.concurrent包的基礎(chǔ)，它提供了一套完整的同步編程框架，開發(fā)人員只需要實現(xiàn)其中幾個簡單的方法就能自由的使用諸如獨(dú)占，共享，條件隊列等多種同步模式。我們常用的比如ReentrantLock，CountDownLatch等等基礎(chǔ)類庫都是基于AQS實現(xiàn)的，足以說明這套框架的強(qiáng)大之處。鑒于此，我們開發(fā)人員更應(yīng)該了解它的實現(xiàn)原理，這樣才能在使用過程中得心應(yīng)手。

總體來說個人感覺AQS的代碼非常難懂，本文就其中的獨(dú)占鎖實現(xiàn)原理進(jìn)行分析。

首先先從整體流程入手，了解下AQS獨(dú)占鎖的執(zhí)行邏輯，然后再一步一步深入分析源碼。

獲取鎖的過程：

當(dāng)線程調(diào)用acquire()申請獲取鎖資源，如果成功，則進(jìn)入臨界區(qū)。

當(dāng)獲取鎖失敗時，則進(jìn)入一個FIFO等待隊列，然后被掛起等待喚醒。

當(dāng)隊列中的等待線程被喚醒以后就重新嘗試獲取鎖資源，如果成功則進(jìn)入臨界區(qū)，否則繼續(xù)掛起等待。

釋放鎖過程：

當(dāng)線程調(diào)用release()進(jìn)行鎖資源釋放時，如果沒有其他線程在等待鎖資源，則釋放完成。

如果隊列中有其他等待鎖資源的線程需要喚醒，則喚醒隊列中的第一個等待節(jié)點(diǎn)（先入先出）。

基于上面所講的獨(dú)占鎖獲取釋放的大致過程，我們再來看下源碼實現(xiàn)邏輯：
首先來看下獲取鎖的方法acquire()

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

代碼雖然短，但包含的邏輯卻很多，一步一步看下：

首先是調(diào)用開發(fā)人員自己實現(xiàn)的tryAcquire() 方法嘗試獲取鎖資源，如果成功則整個acquire()方法執(zhí)行完畢，即當(dāng)前線程獲得鎖資源，可以進(jìn)入臨界區(qū)。

如果獲取鎖失敗，則開始進(jìn)入后面的邏輯，首先是addWaiter(Node.EXCLUSIVE)方法。來看下這個方法的源碼實現(xiàn)：

    //注意：該入隊方法的返回值就是新創(chuàng)建的節(jié)點(diǎn)
    private Node addWaiter(Node mode) {
        //基于當(dāng)前線程，節(jié)點(diǎn)類型（Node.EXCLUSIVE）創(chuàng)建新的節(jié)點(diǎn)
        //由于這里是獨(dú)占模式，因此節(jié)點(diǎn)類型就是Node.EXCLUSIVE
        Node node = new Node(Thread.currentThread(), mode);
        Node pred = tail;
        //這里為了提搞性能，首先執(zhí)行一次快速入隊操作，即直接嘗試將新節(jié)點(diǎn)加入隊尾
        if (pred != null) {
            node.prev = pred;
            //這里根據(jù)CAS的邏輯，即使并發(fā)操作也只能有一個線程成功并返回，其余的都要執(zhí)行后面的入隊操作。即enq()方法
            if (compareAndSetTail(pred, node)) {
                pred.next = node;
                return node;
            }
        }
        enq(node);
        return node;
    }

    //完整的入隊操作
    private Node enq(final Node node) {
        for (;;) {
            Node t = tail;
            //如果隊列還沒有初始化，則進(jìn)行初始化，即創(chuàng)建一個空的頭節(jié)點(diǎn)
            if (t == null) { 
                //同樣是CAS，只有一個線程可以初始化頭結(jié)點(diǎn)成功，其余的都要重復(fù)執(zhí)行循環(huán)體
                if (compareAndSetHead(new Node()))
                    tail = head;
            } else {
                //新創(chuàng)建的節(jié)點(diǎn)指向隊列尾節(jié)點(diǎn)，毫無疑問并發(fā)情況下這里會有多個新創(chuàng)建的節(jié)點(diǎn)指向隊列尾節(jié)點(diǎn)
                node.prev = t;
                //基于這一步的CAS，不管前一步有多少新節(jié)點(diǎn)都指向了尾節(jié)點(diǎn)，這一步只有一個能真正入隊成功，其他的都必須重新執(zhí)行循環(huán)體
                if (compareAndSetTail(t, node)) {
                    t.next = node;
                    //該循環(huán)體唯一退出的操作，就是入隊成功（否則就要無限重試）
                    return t;
                }
            }
        }
    }

上面的入隊操作有兩點(diǎn)需要說明：
一、初始化隊列的觸發(fā)條件就是當(dāng)前已經(jīng)有線程占有了鎖資源，因此上面創(chuàng)建的空的頭節(jié)點(diǎn)可以認(rèn)為就是當(dāng)前占有鎖資源的節(jié)點(diǎn)（雖然它并沒有設(shè)置任何屬性）。
二、注意整個代碼是處在一個死循環(huán)中，知道入隊成功。如果失敗了就會不斷進(jìn)行重試。

經(jīng)過上面的操作，我們申請獲取鎖的線程已經(jīng)成功加入了等待隊列，通過文章最一開始說的獨(dú)占鎖獲取流程，那么節(jié)點(diǎn)現(xiàn)在要做的就是掛起當(dāng)前線程，等待被喚醒，這個邏輯是怎么實現(xiàn)的呢？來看下源碼：

通過上面的分析，該方法入?yún)ode就是剛?cè)腙牭陌?dāng)前線程信息的節(jié)點(diǎn)
final boolean acquireQueued(final Node node, int arg) {
        //鎖資源獲取失敗標(biāo)記位
        boolean failed = true;
        try {
            //等待線程被中斷標(biāo)記位
            boolean interrupted = false;
            //這個循環(huán)體執(zhí)行的時機(jī)包括新節(jié)點(diǎn)入隊和隊列中等待節(jié)點(diǎn)被喚醒兩個地方
            for (;;) {
                //獲取當(dāng)前節(jié)點(diǎn)的前置節(jié)點(diǎn)
                final Node p = node.predecessor();
                //如果前置節(jié)點(diǎn)就是頭結(jié)點(diǎn)，則嘗試獲取鎖資源
                if (p == head && tryAcquire(arg)) {
                    //當(dāng)前節(jié)點(diǎn)獲得鎖資源以后設(shè)置為頭節(jié)點(diǎn)，這里繼續(xù)理解我上面說的那句話
                    //頭結(jié)點(diǎn)就表示當(dāng)前正占有鎖資源的節(jié)點(diǎn)
                    setHead(node);
                    p.next = null; //幫助GC
                    //表示鎖資源成功獲取，因此把failed置為false
                    failed = false;
                    //返回中斷標(biāo)記，表示當(dāng)前節(jié)點(diǎn)是被正常喚醒還是被中斷喚醒
                    return interrupted;
                }
                如果沒有獲取鎖成功，則進(jìn)入掛起邏輯
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
            //最后會分析獲取鎖失敗處理邏輯
            if (failed)
                cancelAcquire(node);
        }
    }

掛起邏輯是很重要的邏輯，這里拿出來多帶帶分析一下，首先要注意目前為止，我們只是根據(jù)當(dāng)前線程，節(jié)點(diǎn)類型創(chuàng)建了一個節(jié)點(diǎn)并加入隊列中，其他屬性都是默認(rèn)值。

//首先說明一下參數(shù)，node是當(dāng)前線程的節(jié)點(diǎn)，pred是它的前置節(jié)點(diǎn)
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        //獲取前置節(jié)點(diǎn)的waitStatus
        int ws = pred.waitStatus;
        if (ws == Node.SIGNAL)
            //如果前置節(jié)點(diǎn)的waitStatus是Node.SIGNAL則返回true，然后會執(zhí)行parkAndCheckInterrupt()方法進(jìn)行掛起
            return true;
        if (ws > 0) {
            //由waitStatus的幾個取值可以判斷這里表示前置節(jié)點(diǎn)被取消
            do {
                node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0);
            //這里我們由當(dāng)前節(jié)點(diǎn)的前置節(jié)點(diǎn)開始，一直向前找最近的一個沒有被取消的節(jié)點(diǎn)
            //注，由于頭結(jié)點(diǎn)head是通過new Node()創(chuàng)建，它的waitStatus為0,因此這里不會出現(xiàn)空指針問題，也就是說最多就是找到頭節(jié)點(diǎn)上面的循環(huán)就退出了
            pred.next = node;
        } else {
            //根據(jù)waitStatus的取值限定，這里waitStatus的值只能是0或者PROPAGATE，那么我們把前置節(jié)點(diǎn)的waitStatus設(shè)為Node.SIGNAL然后重新進(jìn)入該方法進(jìn)行判斷
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        return false;
    }

上面這個方法邏輯比較復(fù)雜，它是用來判斷當(dāng)前節(jié)點(diǎn)是否可以被掛起，也就是喚醒條件是否已經(jīng)具備，即如果掛起了，那一定是可以由其他線程來喚醒的。該方法如果返回false，即掛起條件沒有完備，那就會重新執(zhí)行acquireQueued方法的循環(huán)體，進(jìn)行重新判斷，如果返回true，那就表示萬事俱備，可以掛起了，就會進(jìn)入parkAndCheckInterrupt()方法看下源碼：

private final boolean parkAndCheckInterrupt() {
        LockSupport.park(this);
        //被喚醒之后，返回中斷標(biāo)記，即如果是正常喚醒則返回false，如果是由于中斷醒來，就返回true
        return Thread.interrupted();
    }

看acquireQueued方法中的源碼，如果是因為中斷醒來，那么就把中斷標(biāo)記置為true。不管是正常被喚醒還是由與中斷醒來，都會去嘗試獲取鎖資源。如果成功則返回中斷標(biāo)記，否則繼續(xù)掛起等待。
注：Thread.interrupted()方法在返回中斷標(biāo)記的同時會清除中斷標(biāo)記，也就是說當(dāng)由于中斷醒來然后獲取鎖成功，那么整個acquireQueued方法就會返回true表示是因為中斷醒來，但如果中斷醒來以后沒有獲取到鎖，繼續(xù)掛起，由于這次的中斷已經(jīng)被清除了，下次如果是被正常喚醒，那么acquireQueued方法就會返回false，表示沒有中斷。

最后我們回到acquireQueued方法的最后一步，finally模塊。這里是針對鎖資源獲取失敗以后做的一些善后工作，翻看上面的代碼，其實能進(jìn)入這里的就是tryAcquire()方法拋出異常，也就是說AQS框架針對開發(fā)人員自己實現(xiàn)的獲取鎖操作如果拋出異常，也做了妥善的處理，一起來看下源碼：

//傳入的方法參數(shù)是當(dāng)前獲取鎖資源失敗的節(jié)點(diǎn)
private void cancelAcquire(Node node) {
        // 如果節(jié)點(diǎn)不存在則直接忽略
        if (node == null)
            return;
        
        node.thread = null;

        // 跳過所有已經(jīng)取消的前置節(jié)點(diǎn)，跟上面的那段跳轉(zhuǎn)邏輯類似
        Node pred = node.prev;
        while (pred.waitStatus > 0)
            node.prev = pred = pred.prev;
        //這個是前置節(jié)點(diǎn)的后繼節(jié)點(diǎn)，由于上面可能的跳節(jié)點(diǎn)的操作，所以這里可不一定就是當(dāng)前節(jié)點(diǎn)，仔細(xì)想一下。^_^
        Node predNext = pred.next;

        //把當(dāng)前節(jié)點(diǎn)waitStatus置為取消，這樣別的節(jié)點(diǎn)在處理時就會跳過該節(jié)點(diǎn)
        node.waitStatus = Node.CANCELLED;
        //如果當(dāng)前是尾節(jié)點(diǎn)，則直接刪除，即出隊
        //注：這里不用關(guān)心CAS失敗，因為即使并發(fā)導(dǎo)致失敗，該節(jié)點(diǎn)也已經(jīng)被成功刪除
        if (node == tail && compareAndSetTail(node, pred)) {
            compareAndSetNext(pred, predNext, null);
        } else {
            int ws;
            if (pred != head &&
                ((ws = pred.waitStatus) == Node.SIGNAL ||
                 (ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) &&
                pred.thread != null) {
                Node next = node.next;
                if (next != null && next.waitStatus <= 0)
                    //這里的判斷邏輯很繞，具體就是如果當(dāng)前節(jié)點(diǎn)的前置節(jié)點(diǎn)不是頭節(jié)點(diǎn)且它后面的節(jié)點(diǎn)等待它喚醒（waitStatus小于0），
                    //再加上如果當(dāng)前節(jié)點(diǎn)的后繼節(jié)點(diǎn)沒有被取消就把前置節(jié)點(diǎn)跟后置節(jié)點(diǎn)進(jìn)行連接，相當(dāng)于刪除了當(dāng)前節(jié)點(diǎn)
                    compareAndSetNext(pred, predNext, next);
            } else {
                //進(jìn)入這里，要么當(dāng)前節(jié)點(diǎn)的前置節(jié)點(diǎn)是頭結(jié)點(diǎn)，要么前置節(jié)點(diǎn)的waitStatus是PROPAGATE，直接喚醒當(dāng)前節(jié)點(diǎn)的后繼節(jié)點(diǎn)
                unparkSuccessor(node);
            }

            node.next = node; // help GC
        }
    }

上面就是獨(dú)占模式獲取鎖的核心源碼，確實非常難懂，很繞，就這幾個方法需要反反復(fù)復(fù)看很多遍，才能慢慢理解。

接下來看下釋放鎖的過程：

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

tryRelease()方法是用戶自定義的釋放鎖邏輯，如果成功，就判斷等待隊列中有沒有需要被喚醒的節(jié)點(diǎn)（waitStatus為0表示沒有需要被喚醒的節(jié)點(diǎn)），一起看下喚醒操作：

private void unparkSuccessor(Node node) {
        int ws = node.waitStatus;
        if (ws < 0)
            //把標(biāo)記為設(shè)置為0，表示喚醒操作已經(jīng)開始進(jìn)行，提高并發(fā)環(huán)境下性能
            compareAndSetWaitStatus(node, ws, 0);

        Node s = node.next;
        //如果當(dāng)前節(jié)點(diǎn)的后繼節(jié)點(diǎn)為null，或者已經(jīng)被取消
        if (s == null || s.waitStatus > 0) {
            s = null;
            //注意這個循環(huán)沒有break，也就是說它是從后往前找，一直找到離當(dāng)前節(jié)點(diǎn)最近的一個等待喚醒的節(jié)點(diǎn)
            for (Node t = tail; t != null && t != node; t = t.prev)
                if (t.waitStatus <= 0)
                    s = t;
        }
        //執(zhí)行喚醒操作
        if (s != null)
            LockSupport.unpark(s.thread);
    }

相比而言，鎖的釋放操作就簡單很多了，代碼也比較少。

以上就是AQS獨(dú)占鎖的獲取與釋放過程，大致思想很簡單，就是嘗試去獲取鎖，如果失敗就加入一個隊列中掛起。釋放鎖時，如果隊列中有等待的線程就進(jìn)行喚醒。但如果一步一步看源碼，會發(fā)現(xiàn)細(xì)節(jié)非常多，很多地方很難搞明白，我自己也是反反復(fù)復(fù)學(xué)習(xí)很久才有點(diǎn)心得，但也不敢說已經(jīng)研究通了AQS，甚至不敢說我上面的研究成果就是對的，只是寫篇文章總結(jié)一下，跟同行交流交流心得。
除了獨(dú)占鎖，后面還會產(chǎn)出AQS一系列的文章，包括共享鎖，條件隊列的實現(xiàn)原理等。