摘要：簡(jiǎn)介抽象隊(duì)列同步器，以下簡(jiǎn)稱出現(xiàn)在中，由大師所創(chuàng)作。獲取成功則返回，獲取失敗，線程進(jìn)入同步隊(duì)列等待。響應(yīng)中斷版的超時(shí)響應(yīng)中斷版的共享式獲取同步狀態(tài)，同一時(shí)刻可能會(huì)有多個(gè)線程獲得同步狀態(tài)。

AbstractQueuedSynchronizer （抽象隊(duì)列同步器，以下簡(jiǎn)稱 AQS）出現(xiàn)在 JDK 1.5 中，由大師 Doug Lea 所創(chuàng)作。AQS 是很多同步器的基礎(chǔ)框架，比如 ReentrantLock、CountDownLatch 和 Semaphore 等都是基于 AQS 實(shí)現(xiàn)的。除此之外，我們還可以基于 AQS，定制出我們所需要的同步器。

AQS 的使用方式通常都是通過(guò)內(nèi)部類繼承 AQS 實(shí)現(xiàn)同步功能，通過(guò)繼承 AQS，可以簡(jiǎn)化同步器的實(shí)現(xiàn)。如前面所說(shuō)，AQS 是很多同步器實(shí)現(xiàn)的基礎(chǔ)框架。弄懂 AQS 對(duì)理解 Java 并發(fā)包里的組件大有裨益，這也是我學(xué)習(xí) AQS 并寫出這篇文章的緣由。另外，需要說(shuō)明的是，AQS 本身并不是很好理解，細(xì)節(jié)很多。在看的過(guò)程中藥有一定的耐心，做好看多遍的準(zhǔn)備。好了，其他的就不多說(shuō)了，開始進(jìn)入正題吧。

在 AQS 內(nèi)部，通過(guò)維護(hù)一個(gè)FIFO 隊(duì)列來(lái)管理多線程的排隊(duì)工作。在公平競(jìng)爭(zhēng)的情況下，無(wú)法獲取同步狀態(tài)的線程將會(huì)被封裝成一個(gè)節(jié)點(diǎn)，置于隊(duì)列尾部。入隊(duì)的線程將會(huì)通過(guò)自旋的方式獲取同步狀態(tài)，若在有限次的嘗試后，仍未獲取成功，線程則會(huì)被阻塞住。大致示意圖如下：

當(dāng)頭結(jié)點(diǎn)釋放同步狀態(tài)后，且后繼節(jié)點(diǎn)對(duì)應(yīng)的線程被阻塞，此時(shí)頭結(jié)點(diǎn)

線程將會(huì)去喚醒后繼節(jié)點(diǎn)線程。后繼節(jié)點(diǎn)線程恢復(fù)運(yùn)行并獲取同步狀態(tài)后，會(huì)將舊的頭結(jié)點(diǎn)從隊(duì)列中移除，并將自己設(shè)為頭結(jié)點(diǎn)。大致示意圖如下：

本節(jié)將介紹三組重要的方法，通過(guò)使用這三組方法即可實(shí)現(xiàn)一個(gè)同步組件。

第一組方法是用于訪問(wèn)/設(shè)置同步狀態(tài)的，如下：

第二組方需要由同步組件覆寫。如下：

第三組方法是一組模板方法，同步組件可直接調(diào)用。如下：

上面列舉了一堆方法，看似繁雜。但稍微理一下，就會(huì)發(fā)現(xiàn)上面諸多方法無(wú)非就兩大類：一類是獨(dú)占式獲取和釋放共享狀態(tài)，另一類是共享式獲取和釋放同步狀態(tài)。至于這兩類方法的實(shí)現(xiàn)細(xì)節(jié)，我會(huì)在接下來(lái)的章節(jié)中講到，繼續(xù)往下看吧。

在并發(fā)的情況下，AQS 會(huì)將未獲取同步狀態(tài)的線程將會(huì)封裝成節(jié)點(diǎn)，并將其放入同步隊(duì)列尾部。同步隊(duì)列中的節(jié)點(diǎn)除了要保存線程，還要保存等待狀態(tài)。不管是獨(dú)占式還是共享式，在獲取狀態(tài)失敗時(shí)都會(huì)用到節(jié)點(diǎn)類。所以這里我們要先看一下節(jié)點(diǎn)類的實(shí)現(xiàn)，為后面的源碼分析進(jìn)行簡(jiǎn)單鋪墊。源碼如下：

static final class Node {

    /** 共享類型節(jié)點(diǎn)，標(biāo)記節(jié)點(diǎn)在共享模式下等待 */
    static final Node SHARED = new Node();
    
    /** 獨(dú)占類型節(jié)點(diǎn)，標(biāo)記節(jié)點(diǎn)在獨(dú)占模式下等待 */
    static final Node EXCLUSIVE = null;

    /** 等待狀態(tài) - 取消 */
    static final int CANCELLED =  1;
    
    /** 
     * 等待狀態(tài) - 通知。某個(gè)節(jié)點(diǎn)是處于該狀態(tài)，當(dāng)該節(jié)點(diǎn)釋放同步狀態(tài)后，
     * 會(huì)通知后繼節(jié)點(diǎn)線程，使之可以恢復(fù)運(yùn)行 
     */
    static final int SIGNAL    = -1;
    
    /** 等待狀態(tài) - 條件等待。表明節(jié)點(diǎn)等待在 Condition 上 */
    static final int CONDITION = -2;
    
    /**
     * 等待狀態(tài) - 傳播。表示無(wú)條件向后傳播喚醒動(dòng)作，詳細(xì)分析請(qǐng)看第五章
     */
    static final int PROPAGATE = -3;

    /**
     * 等待狀態(tài)，取值如下：
     *   SIGNAL,
     *   CANCELLED,
     *   CONDITION,
     *   PROPAGATE,
     *   0
     * 
     * 初始情況下，waitStatus = 0
     */
    volatile int waitStatus;

    /**
     * 前驅(qū)節(jié)點(diǎn)
     */
    volatile Node prev;

    /**
     * 后繼節(jié)點(diǎn)
     */
    volatile Node next;

    /**
     * 對(duì)應(yīng)的線程
     */
    volatile Thread thread;

    /**
     * 下一個(gè)等待節(jié)點(diǎn)，用在 ConditionObject 中
     */
    Node nextWaiter;

    /**
     * 判斷節(jié)點(diǎn)是否是共享節(jié)點(diǎn)
     */
    final boolean isShared() {
        return nextWaiter == SHARED;
    }

    /**
     * 獲取前驅(qū)節(jié)點(diǎn)
     */
    final Node predecessor() throws NullPointerException {
        Node p = prev;
        if (p == null)
            throw new NullPointerException();
        else
            return p;
    }

    Node() {    // Used to establish initial head or SHARED marker
    }

    /** addWaiter 方法會(huì)調(diào)用該構(gòu)造方法 */
    Node(Thread thread, Node mode) {
        this.nextWaiter = mode;
        this.thread = thread;
    }

    /** Condition 中會(huì)用到此構(gòu)造方法 */
    Node(Thread thread, int waitStatus) { // Used by Condition
        this.waitStatus = waitStatus;
        this.thread = thread;
    }
}

獨(dú)占式獲取同步狀態(tài)時(shí)通過(guò) acquire 進(jìn)行的，下面來(lái)分析一下該方法的源碼。如下：

/**
 * 該方法將會(huì)調(diào)用子類復(fù)寫的 tryAcquire 方法獲取同步狀態(tài)，
 * - 獲取成功：直接返回
 * - 獲取失敗：將線程封裝在節(jié)點(diǎn)中，并將節(jié)點(diǎn)置于同步隊(duì)列尾部，
 *     通過(guò)自旋嘗試獲取同步狀態(tài)。如果在有限次內(nèi)仍無(wú)法獲取同步狀態(tài)，
 *     該線程將會(huì)被 LockSupport.park 方法阻塞住，直到被前驅(qū)節(jié)點(diǎn)喚醒
 */
public final void acquire(int arg) {
    if (!tryAcquire(arg) &&
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
}

/** 向同步隊(duì)列尾部添加一個(gè)節(jié)點(diǎn) */
private Node addWaiter(Node mode) {
    Node node = new Node(Thread.currentThread(), mode);
    // 嘗試以快速方式將節(jié)點(diǎn)添加到隊(duì)列尾部
    Node pred = tail;
    if (pred != null) {
        node.prev = pred;
        if (compareAndSetTail(pred, node)) {
            pred.next = node;
            return node;
        }
    }
    
    // 快速插入節(jié)點(diǎn)失敗，調(diào)用 enq 方法，不停的嘗試插入節(jié)點(diǎn)
    enq(node);
    return node;
}

/**
 * 通過(guò) CAS + 自旋的方式插入節(jié)點(diǎn)到隊(duì)尾
 */
private Node enq(final Node node) {
    for (;;) {
        Node t = tail;
        if (t == null) { // Must initialize
            // 設(shè)置頭結(jié)點(diǎn)，初始情況下，頭結(jié)點(diǎn)是一個(gè)空節(jié)點(diǎn)
            if (compareAndSetHead(new Node()))
                tail = head;
        } else {
            /*
             * 將節(jié)點(diǎn)插入隊(duì)列尾部。這里是先將新節(jié)點(diǎn)的前驅(qū)設(shè)為尾節(jié)點(diǎn)，之后在嘗試將新節(jié)點(diǎn)設(shè)為尾節(jié)
             * 點(diǎn)，最后再將原尾節(jié)點(diǎn)的后繼節(jié)點(diǎn)指向新的尾節(jié)點(diǎn)。除了這種方式，我們還先設(shè)置尾節(jié)點(diǎn)，
             * 之后再設(shè)置前驅(qū)和后繼，即：
             * 
             *    if (compareAndSetTail(t, node)) {
             *        node.prev = t;
             *        t.next = node;
             *    }
             *    
             * 但但如果是這樣做，會(huì)導(dǎo)致一個(gè)問(wèn)題，即短時(shí)內(nèi)，隊(duì)列結(jié)構(gòu)會(huì)遭到破壞。考慮這種情況，
             * 某個(gè)線程在調(diào)用 compareAndSetTail(t, node)成功后，該線程被 CPU 切換了。此時(shí)
             * 設(shè)置前驅(qū)和后繼的代碼還沒(méi)帶的及執(zhí)行，但尾節(jié)點(diǎn)指針卻設(shè)置成功，導(dǎo)致隊(duì)列結(jié)構(gòu)短時(shí)內(nèi)會(huì)
             * 出現(xiàn)如下情況：
             *
             *      +------+  prev +-----+       +-----+
             * head |      | <---- |     |       |     |  tail
             *      |      | ----> |     |       |     |
             *      +------+ next  +-----+       +-----+
             *
             * tail 節(jié)點(diǎn)完全脫離了隊(duì)列，這樣導(dǎo)致一些隊(duì)列遍歷代碼出錯(cuò)。如果先設(shè)置
             * 前驅(qū)，在設(shè)置尾節(jié)點(diǎn)。及時(shí)線程被切換，隊(duì)列結(jié)構(gòu)短時(shí)可能如下：
             *
             *      +------+  prev +-----+ prev  +-----+
             * head |      | <---- |     | <---- |     |  tail
             *      |      | ----> |     |       |     |
             *      +------+ next  +-----+       +-----+
             *      
             * 這樣并不會(huì)影響從后向前遍歷，不會(huì)導(dǎo)致遍歷邏輯出錯(cuò)。
             * 
             * 參考：
             *    https://www.cnblogs.com/micrari/p/6937995.html
             */
            node.prev = t;
            if (compareAndSetTail(t, node)) {
                t.next = node;
                return t;
            }
        }
    }
}

/**
 * 同步隊(duì)列中的線程在此方法中以循環(huán)嘗試獲取同步狀態(tài)，在有限次的嘗試后，
 * 若仍未獲取鎖，線程將會(huì)被阻塞，直至被前驅(qū)節(jié)點(diǎn)的線程喚醒。
 */
final boolean acquireQueued(final Node node, int arg) {
    boolean failed = true;
    try {
        boolean interrupted = false;
        // 循環(huán)獲取同步狀態(tài)
        for (;;) {
            final Node p = node.predecessor();
            /*
             * 前驅(qū)節(jié)點(diǎn)如果是頭結(jié)點(diǎn)，表明前驅(qū)節(jié)點(diǎn)已經(jīng)獲取了同步狀態(tài)。前驅(qū)節(jié)點(diǎn)釋放同步狀態(tài)后，
             * 在不出異常的情況下， tryAcquire(arg) 應(yīng)返回 true。此時(shí)節(jié)點(diǎn)就成功獲取了同
             * 步狀態(tài)，并將自己設(shè)為頭節(jié)點(diǎn)，原頭節(jié)點(diǎn)出隊(duì)。
             */ 
            if (p == head && tryAcquire(arg)) {
                // 成功獲取同步狀態(tài)，設(shè)置自己為頭節(jié)點(diǎn)
                setHead(node);
                p.next = null; // help GC
                failed = false;
                return interrupted;
            }
            
            /*
             * 如果獲取同步狀態(tài)失敗，則根據(jù)條件判斷是否應(yīng)該阻塞自己。
             * 如果不阻塞，CPU 就會(huì)處于忙等狀態(tài)，這樣會(huì)浪費(fèi) CPU 資源
             */
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        /*
         * 如果在獲取同步狀態(tài)中出現(xiàn)異常，failed = true，cancelAcquire 方法會(huì)被執(zhí)行。
         * tryAcquire 需同步組件開發(fā)者覆寫，難免不了會(huì)出現(xiàn)異常。
         */
        if (failed)
            cancelAcquire(node);
    }
}

/** 設(shè)置頭節(jié)點(diǎn) */
private void setHead(Node node) {
    // 僅有一個(gè)線程可以成功獲取同步狀態(tài)，所以這里不需要進(jìn)行同步控制
    head = node;
    node.thread = null;
    node.prev = null;
}

/**
 * 該方法主要用途是，當(dāng)線程在獲取同步狀態(tài)失敗時(shí)，根據(jù)前驅(qū)節(jié)點(diǎn)的等待狀態(tài)，決定后續(xù)的動(dòng)作。比如前驅(qū)
 * 節(jié)點(diǎn)等待狀態(tài)為 SIGNAL，表明當(dāng)前節(jié)點(diǎn)線程應(yīng)該被阻塞住了。不能老是嘗試，避免 CPU 忙等。
 *    —————————————————————————————————————————————————————————————————
 *    | 前驅(qū)節(jié)點(diǎn)等待狀態(tài) |                   相應(yīng)動(dòng)作                     |
 *    —————————————————————————————————————————————————————————————————
 *    | SIGNAL         | 阻塞                                          |
 *    | CANCELLED      | 向前遍歷, 移除前面所有為該狀態(tài)的節(jié)點(diǎn)               |
 *    | waitStatus < 0 | 將前驅(qū)節(jié)點(diǎn)狀態(tài)設(shè)為 SIGNAL, 并再次嘗試獲取同步狀態(tài)   |
 *    —————————————————————————————————————————————————————————————————
 */
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    int ws = pred.waitStatus;
    /* 
     * 前驅(qū)節(jié)點(diǎn)等待狀態(tài)為 SIGNAL，表示當(dāng)前線程應(yīng)該被阻塞。
     * 線程阻塞后，會(huì)在前驅(qū)節(jié)點(diǎn)釋放同步狀態(tài)后被前驅(qū)節(jié)點(diǎn)線程喚醒
     */
    if (ws == Node.SIGNAL)
        return true;
        
    /*
     * 前驅(qū)節(jié)點(diǎn)等待狀態(tài)為 CANCELLED，則以前驅(qū)節(jié)點(diǎn)為起點(diǎn)向前遍歷，
     * 移除其他等待狀態(tài)為 CANCELLED 的節(jié)點(diǎn)。
     */ 
    if (ws > 0) {
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;
    } else {
        /*
         * 等待狀態(tài)為 0 或 PROPAGATE，設(shè)置前驅(qū)節(jié)點(diǎn)等待狀態(tài)為 SIGNAL，
         * 并再次嘗試獲取同步狀態(tài)。
         */
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}

private final boolean parkAndCheckInterrupt() {
    // 調(diào)用 LockSupport.park 阻塞自己
    LockSupport.park(this);
    return Thread.interrupted();
}

/**
 * 取消獲取同步狀態(tài)
 */
private void cancelAcquire(Node node) {
    if (node == null)
        return;

    node.thread = null;

    // 前驅(qū)節(jié)點(diǎn)等待狀態(tài)為 CANCELLED，則向前遍歷并移除其他為該狀態(tài)的節(jié)點(diǎn)
    Node pred = node.prev;
    while (pred.waitStatus > 0)
        node.prev = pred = pred.prev;

    // 記錄 pred 的后繼節(jié)點(diǎn)，后面會(huì)用到
    Node predNext = pred.next;

    // 將當(dāng)前節(jié)點(diǎn)等待狀態(tài)設(shè)為 CANCELLED
    node.waitStatus = Node.CANCELLED;

    /*
     * 如果當(dāng)前節(jié)點(diǎn)是尾節(jié)點(diǎn)，則通過(guò) CAS 設(shè)置前驅(qū)節(jié)點(diǎn) prev 為尾節(jié)點(diǎn)。設(shè)置成功后，再利用 CAS 將 
     * prev 的 next 引用置空，斷開與后繼節(jié)點(diǎn)的聯(lián)系，完成清理工作。
     */ 
    if (node == tail && compareAndSetTail(node, pred)) {
        /* 
         * 執(zhí)行到這里，表明 pred 節(jié)點(diǎn)被成功設(shè)為了尾節(jié)點(diǎn)，這里通過(guò) CAS 將 pred 節(jié)點(diǎn)的后繼節(jié)點(diǎn)
         * 設(shè)為 null。注意這里的 CAS 即使失敗了，也沒(méi)關(guān)系。失敗了，表明 pred 的后繼節(jié)點(diǎn)更新
         * 了。pred 此時(shí)已經(jīng)是尾節(jié)點(diǎn)了，若后繼節(jié)點(diǎn)被更新，則是有新節(jié)點(diǎn)入隊(duì)了。這種情況下，CAS 
         * 會(huì)失敗，但失敗不會(huì)影響同步隊(duì)列的結(jié)構(gòu)。
         */
        compareAndSetNext(pred, predNext, null);
    } else {
        int ws;
        // 根據(jù)條件判斷是喚醒后繼節(jié)點(diǎn)，還是將前驅(qū)節(jié)點(diǎn)和后繼節(jié)點(diǎn)連接到一起
        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)
                /*
                 * 這里使用 CAS 設(shè)置 pred 的 next，表明多個(gè)線程同時(shí)在取消，這里存在競(jìng)爭(zhēng)。
                 * 不過(guò)此處沒(méi)針對(duì) compareAndSetNext 方法失敗后做一些處理，表明即使失敗了也
                 * 沒(méi)關(guān)系。實(shí)際上，多個(gè)線程同時(shí)設(shè)置 pred 的 next 引用時(shí)，只要有一個(gè)能設(shè)置成
                 * 功即可。
                 */
                compareAndSetNext(pred, predNext, next);
        } else {
            /*
             * 喚醒后繼節(jié)點(diǎn)對(duì)應(yīng)的線程。這里簡(jiǎn)單講一下為什么要喚醒后繼線程，考慮下面一種情況：
             *        head          node1         node2         tail
             *        ws=0          ws=1          ws=-1         ws=0
             *      +------+  prev +-----+  prev +-----+  prev +-----+
             *      |      | <---- |     | <---- |     | <---- |     |  
             *      |      | ----> |     | ----> |     | ----> |     |
             *      +------+  next +-----+  next +-----+  next +-----+
             *      
             * 頭結(jié)點(diǎn)初始狀態(tài)為 0，node1、node2 和 tail 節(jié)點(diǎn)依次入隊(duì)。node1 自旋過(guò)程中調(diào)用 
             * tryAcquire 出現(xiàn)異常，進(jìn)入 cancelAcquire。head 節(jié)點(diǎn)此時(shí)等待狀態(tài)仍然是 0，它
             * 會(huì)認(rèn)為后繼節(jié)點(diǎn)還在運(yùn)行中，所它在釋放同步狀態(tài)后，不會(huì)去喚醒后繼等待狀態(tài)為非取消的
             * 節(jié)點(diǎn) node2。如果 node1 再不喚醒 node2 的線程，該線程面臨無(wú)法被喚醒的情況。此
             * 時(shí)，整個(gè)同步隊(duì)列就回全部阻塞住。
             */
            unparkSuccessor(node);
        }

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

private void unparkSuccessor(Node node) {
    int ws = node.waitStatus;
    /*
     * 通過(guò) CAS 將等待狀態(tài)設(shè)為 0，讓后繼節(jié)點(diǎn)線程多一次
     * 嘗試獲取同步狀態(tài)的機(jī)會(huì)
     */
    if (ws < 0)
        compareAndSetWaitStatus(node, ws, 0);

    Node s = node.next;
    if (s == null || s.waitStatus > 0) {
        s = null;
       /*
        * 這里如果 s == null 處理，是不是表明 node 是尾節(jié)點(diǎn)？答案是不一定。原因之前在分析 
        * enq 方法時(shí)說(shuō)過(guò)。這里再啰嗦一遍，新節(jié)點(diǎn)入隊(duì)時(shí)，隊(duì)列瞬時(shí)結(jié)構(gòu)可能如下：
        *                      node1         node2
        *      +------+  prev +-----+ prev  +-----+
        * head |      | <---- |     | <---- |     |  tail
        *      |      | ----> |     |       |     |
        *      +------+ next  +-----+       +-----+
        * 
        * node2 節(jié)點(diǎn)為新入隊(duì)節(jié)點(diǎn)，此時(shí) tail 已經(jīng)指向了它，但 node1 后繼引用還未設(shè)置。
        * 這里 node1 就是 node 參數(shù)，s = node1.next = null，但此時(shí) node1 并不是尾
        * 節(jié)點(diǎn)。所以這里不能從前向后遍歷同步隊(duì)列，應(yīng)該從后向前。
        */
        for (Node t = tail; t != null && t != node; t = t.prev)
            if (t.waitStatus <= 0)
                s = t;
    }
    if (s != null)
        // 喚醒 node 的后繼節(jié)點(diǎn)線程
        LockSupport.unpark(s.thread);
}

到這里，獨(dú)占式獲取同步狀態(tài)的分析就講完了。如果僅分析獲取同步狀態(tài)的大致流程，那么這個(gè)流程并不難。但若深入到細(xì)節(jié)之中，還是需要思考思考。這里對(duì)獨(dú)占式獲取同步狀態(tài)的大致流程做個(gè)總結(jié)，如下：

調(diào)用 tryAcquire 方法嘗試獲取同步狀態(tài)

獲取成功，直接返回

獲取失敗，將線程封裝到節(jié)點(diǎn)中，并將節(jié)點(diǎn)入隊(duì)

入隊(duì)節(jié)點(diǎn)在 acquireQueued 方法中自旋獲取同步狀態(tài)

若節(jié)點(diǎn)的前驅(qū)節(jié)點(diǎn)是頭節(jié)點(diǎn)，則再次調(diào)用 tryAcquire 嘗試獲取同步狀態(tài)

獲取成功，當(dāng)前節(jié)點(diǎn)將自己設(shè)為頭節(jié)點(diǎn)并返回

獲取失敗，可能再次嘗試，也可能會(huì)被阻塞。這里簡(jiǎn)單認(rèn)為會(huì)被阻塞。

上面的步驟對(duì)應(yīng)下面的流程圖：

上面流程圖參考自《Java并發(fā)編程》第128頁(yè)圖 5-5，這里進(jìn)行了重新繪制，并做了一定的修改。

相對(duì)于獲取同步狀態(tài)，釋放同步狀態(tài)的過(guò)程則要簡(jiǎn)單的多，這里簡(jiǎn)單羅列一下步驟：

調(diào)用 tryRelease(arg) 嘗試釋放同步狀態(tài)

根據(jù)條件判斷是否應(yīng)該喚醒后繼線程

就兩個(gè)步驟，下面看一下源碼分析。

public final boolean release(int arg) {
    if (tryRelease(arg)) {
        Node h = head;
        /*
         * 這里簡(jiǎn)單列舉條件分支的可能性，如下：
         * 1. head = null
         *     head 還未初始化。初始情況下，head = null，當(dāng)?shù)谝粋€(gè)節(jié)點(diǎn)入隊(duì)后，head 會(huì)被初始
         *     為一個(gè)虛擬（dummy）節(jié)點(diǎn)。這里，如果還沒(méi)節(jié)點(diǎn)入隊(duì)就調(diào)用 release 釋放同步狀態(tài)，
         *     就會(huì)出現(xiàn) h = null 的情況。
         *     
         * 2. head != null && waitStatus = 0
         *     表明后繼節(jié)點(diǎn)對(duì)應(yīng)的線程仍在運(yùn)行中，不需要喚醒
         * 
         * 3. head != null && waitStatus < 0
         *     后繼節(jié)點(diǎn)對(duì)應(yīng)的線程可能被阻塞了，需要喚醒 
         */
        if (h != null && h.waitStatus != 0)
            // 喚醒后繼節(jié)點(diǎn)，上面分析過(guò)了，這里不再贅述
            unparkSuccessor(h);
        return true;
    }
    return false;
}

與獨(dú)占模式不同，共享模式下，同一時(shí)刻會(huì)有多個(gè)線程獲取共享同步狀態(tài)。共享模式是實(shí)現(xiàn)讀寫鎖中的讀鎖、CountDownLatch 和 Semaphore 等同步組件的基礎(chǔ)，搞懂了，再去理解一些共享同步組件就不難了。

共享類型的節(jié)點(diǎn)獲取共享同步狀態(tài)后，如果后繼節(jié)點(diǎn)也是共享類型節(jié)點(diǎn)，當(dāng)前節(jié)點(diǎn)則會(huì)喚醒后繼節(jié)點(diǎn)。這樣，多個(gè)節(jié)點(diǎn)線程即可同時(shí)獲取共享同步狀態(tài)。

public final void acquireShared(int arg) {
    // 嘗試獲取共享同步狀態(tài)，tryAcquireShared 返回的是整型
    if (tryAcquireShared(arg) < 0)
        doAcquireShared(arg);
}

private void doAcquireShared(int arg) {
    final Node node = addWaiter(Node.SHARED);
    boolean failed = true;
    try {
        boolean interrupted = false;
        // 這里和前面一樣，也是通過(guò)有限次自旋的方式獲取同步狀態(tài)
        for (;;) {
            final Node p = node.predecessor();
            /*
             * 前驅(qū)是頭結(jié)點(diǎn)，其類型可能是 EXCLUSIVE，也可能是 SHARED.
             * 如果是 EXCLUSIVE，線程無(wú)法獲取共享同步狀態(tài)。
             * 如果是 SHARED，線程則可獲取共享同步狀態(tài)。
             * 能不能獲取共享同步狀態(tài)要看 tryAcquireShared 具體的實(shí)現(xiàn)。比如多個(gè)線程競(jìng)爭(zhēng)讀寫
             * 鎖的中的讀鎖時(shí)，均能成功獲取讀鎖。但多個(gè)線程同時(shí)競(jìng)爭(zhēng)信號(hào)量時(shí)，可能就會(huì)有一部分線
             * 程因無(wú)法競(jìng)爭(zhēng)到信號(hào)量資源而阻塞。
             */ 
            if (p == head) {
                // 嘗試獲取共享同步狀態(tài)
                int r = tryAcquireShared(arg);
                if (r >= 0) {
                    // 設(shè)置頭結(jié)點(diǎn)，如果后繼節(jié)點(diǎn)是共享類型，喚醒后繼節(jié)點(diǎn)
                    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);
    }
}
   
/**
 * 這個(gè)方法做了兩件事情：
 * 1. 設(shè)置自身為頭結(jié)點(diǎn)
 * 2. 根據(jù)條件判斷是否要喚醒后繼節(jié)點(diǎn)
 */ 
private void setHeadAndPropagate(Node node, int propagate) {
    Node h = head;
    // 設(shè)置頭結(jié)點(diǎn)
    setHead(node);
    
    /*
     * 這個(gè)條件分支由 propagate > 0 和 h.waitStatus < 0 兩部分組成。
     * h.waitStatus < 0 時(shí)，waitStatus = SIGNAL 或 PROPAGATE。這里僅依賴
     * 條件 propagate > 0 判斷是否喚醒后繼節(jié)點(diǎn)是不充分的，至于原因請(qǐng)參考第五章
     */
    if (propagate > 0 || h == null || h.waitStatus < 0 ||
        (h = head) == null || h.waitStatus < 0) {
        Node s = node.next;
        /*
         * 節(jié)點(diǎn) s 如果是共享類型節(jié)點(diǎn)，則應(yīng)該喚醒該節(jié)點(diǎn)
         * 至于 s == null 的情況前面分析過(guò)，這里不在贅述。
         */ 
        if (s == null || s.isShared())
            doReleaseShared();
    }
}

/**
 * 該方法用于在 acquires/releases 存在競(jìng)爭(zhēng)的情況下，確保喚醒動(dòng)作向后傳播。
 */ 
private void doReleaseShared() {
    /*
     * 下面的循環(huán)在 head 節(jié)點(diǎn)存在后繼節(jié)點(diǎn)的情況下，做了兩件事情：
     * 1. 如果 head 節(jié)點(diǎn)等待狀態(tài)為 SIGNAL，則將 head 節(jié)點(diǎn)狀態(tài)設(shè)為 0，并喚醒后繼節(jié)點(diǎn)
     * 2. 如果 head 節(jié)點(diǎn)等待狀態(tài)為 0，則將 head 節(jié)點(diǎn)狀態(tài)設(shè)為 PROPAGATE，保證喚醒能夠正
     *    常傳播下去。關(guān)于 PROPAGATE 狀態(tài)的細(xì)節(jié)分析，后面會(huì)講到。
     */
    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);
            }
            /* 
             * ws = 0 的情況下，這里要嘗試將狀態(tài)從 0 設(shè)為 PROPAGATE，保證喚醒向后
             * 傳播。setHeadAndPropagate 在讀到 h.waitStatus < 0 時(shí)，可以繼續(xù)喚醒
             * 后面的節(jié)點(diǎn)。
             */
            else if (ws == 0 &&
                     !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                continue;                // loop on failed CAS
        }
        if (h == head)                   // loop if head changed
            break;
    }
}

到這里，共享模式下獲取同步狀態(tài)的邏輯就分析完了，不過(guò)我這里只做了簡(jiǎn)單分析。相對(duì)于獨(dú)占式獲取同步狀態(tài)，共享式的情況更為復(fù)雜。獨(dú)占模式下，只有一個(gè)節(jié)點(diǎn)線程可以成功獲取同步狀態(tài)，也只有獲取已同步狀態(tài)節(jié)點(diǎn)線程才可以釋放同步狀態(tài)。但在共享模式下，多個(gè)共享節(jié)點(diǎn)線程可以同時(shí)獲得同步狀態(tài)，在一些線程獲取同步狀態(tài)的同時(shí)，可能還會(huì)有另外一些線程正在釋放同步狀態(tài)。所以，共享模式更為復(fù)雜。這里我的腦力跟不上了，沒(méi)法面面俱到的分析，見(jiàn)諒。

最后說(shuō)一下共享模式下獲取同步狀態(tài)的大致流程，如下：

獲取共享同步狀態(tài)

若獲取失敗，則生成節(jié)點(diǎn)，并入隊(duì)

如果前驅(qū)為頭結(jié)點(diǎn)，再次嘗試獲取共享同步狀態(tài)

獲取成功則將自己設(shè)為頭結(jié)點(diǎn)，如果后繼節(jié)點(diǎn)是共享類型的，則喚醒

若失敗，將節(jié)點(diǎn)狀態(tài)設(shè)為 SIGNAL，再次嘗試。若再次失敗，線程進(jìn)入等待狀態(tài)

釋放共享狀態(tài)主要邏輯在 doReleaseShared 中，doReleaseShared 上節(jié)已經(jīng)分析過(guò)，這里就不贅述了。共享節(jié)點(diǎn)線程在獲取同步狀態(tài)和釋放同步狀態(tài)時(shí)都會(huì)調(diào)用 doReleaseShared，所以 doReleaseShared 是多線程競(jìng)爭(zhēng)集中的地方。

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

AQS 的節(jié)點(diǎn)有幾種不同的狀態(tài)，這個(gè)在 4.1 節(jié)介紹過(guò)。在這幾個(gè)狀態(tài)中，PROPAGATE 的用途可能是最不好理解的。網(wǎng)上包括一些書籍關(guān)于該狀態(tài)的敘述基本都是一句帶過(guò)，也就是 PROPAGATE 字面意義，即向后傳播喚醒動(dòng)作。至于怎么傳播，鮮有資料說(shuō)明過(guò)。不過(guò)，好在最終我還是找到了一篇詳細(xì)敘述了 PROPAGATE 狀態(tài)的文章。在博客園上，博友 活在夢(mèng)裡 在他的文章 AbstractQueuedSynchronizer源碼解讀 對(duì) PROPAGATE，以及其他的一些細(xì)節(jié)進(jìn)行了說(shuō)明，很有深度。在欽佩之余，不由得感嘆作者思考的很深入。在征得他的同意后，我將在本節(jié)中引用他文章中對(duì) PROPAGATE 狀態(tài)說(shuō)明的部分，并進(jìn)行一定的補(bǔ)充說(shuō)明。這里感謝作者 活在夢(mèng)裡 的精彩分享，若不參考他的文章，我的這篇文章內(nèi)容會(huì)比較空洞。好了，其他的不多說(shuō)了，繼續(xù)往下分析。

在本節(jié)中，將會(huì)說(shuō)明兩個(gè)個(gè)問(wèn)題，如下：

PROPAGATE 狀態(tài)用在哪里，以及怎樣向后傳播喚醒動(dòng)作的？

引入 PROPAGATE 狀態(tài)是為了解決什么問(wèn)題？

這兩個(gè)問(wèn)題將會(huì)在下面兩節(jié)中分別進(jìn)行說(shuō)明。

PROPAGATE 狀態(tài)是用來(lái)傳播喚醒動(dòng)作的，那么它是在哪里進(jìn)行傳播的呢？答案是在setHeadAndPropagate方法中，這里再來(lái)看看 setHeadAndPropagate 方法的實(shí)現(xiàn)：

private void setHeadAndPropagate(Node node, int propagate) {
    Node h = head;
    setHead(node);
    
    if (propagate > 0 || h == null || h.waitStatus < 0 ||
        (h = head) == null || h.waitStatus < 0) {
        Node s = node.next;
        if (s == null || s.isShared())
            doReleaseShared();
    }
}

大家注意看 setHeadAndPropagate 方法中那個(gè)長(zhǎng)長(zhǎng)的判斷語(yǔ)句，其中有一個(gè)條件是h.waitStatus < 0，當(dāng) h.waitStatus = SIGNAL(-1) 或 PROPAGATE(-3) 是，這個(gè)條件就會(huì)成立。那么 PROPAGATE 狀態(tài)是在何時(shí)被設(shè)置的呢？答案是在doReleaseShared方法中，如下：

private void doReleaseShared() {
    for (;;) {
        Node h = head;
        if (h != null && h != tail) {
            int ws = h.waitStatus;
            if (ws == Node.SIGNAL) {...}
            
            // 如果 ws = 0，則將 h 狀態(tài)設(shè)為 PROPAGATE
            else if (ws == 0 &&
                     !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                continue;                // loop on failed CAS
        }
        ...
    }
}

再回到 setHeadAndPropagate 的實(shí)現(xiàn)，該方法既然引入了h.waitStatus < 0這個(gè)條件，就意味著僅靠條件propagate > 0判斷是否喚醒后繼節(jié)點(diǎn)線程的機(jī)制是不充分的。至于為啥不充分，請(qǐng)繼續(xù)往看下看。

PROPAGATE 的引入是為了解決一個(gè) BUG -- JDK-6801020，復(fù)現(xiàn)這個(gè) BUG 的代碼如下：

import java.util.concurrent.Semaphore;

public class TestSemaphore {

   private static Semaphore sem = new Semaphore(0);

   private static class Thread1 extends Thread {
       @Override
       public void run() {
           sem.acquireUninterruptibly();
       }
   }

   private static class Thread2 extends Thread {
       @Override
       public void run() {
           sem.release();
       }
   }

   public static void main(String[] args) throws InterruptedException {
       for (int i = 0; i < 10000000; i++) {
           Thread t1 = new Thread1();
           Thread t2 = new Thread1();
           Thread t3 = new Thread2();
           Thread t4 = new Thread2();
           t1.start();
           t2.start();
           t3.start();
           t4.start();
           t1.join();
           t2.join();
           t3.join();
           t4.join();
           System.out.println(i);
       }
   }
}

根據(jù) BUG 的描述消息可知 JDK 6u11,6u17 兩個(gè)版本受到影響。那么，接下來(lái)再來(lái)看看引起這個(gè) BUG 的代碼 -- JDK 6u17 中 setHeadAndPropagate 和 releaseShared 兩個(gè)方法源碼，如下：

private void setHeadAndPropagate(Node node, int propagate) {
    setHead(node);
    if (propagate > 0 && node.waitStatus != 0) {
        /*
         * Don"t bother fully figuring out successor.  If it
         * looks null, call unparkSuccessor anyway to be safe.
         */
        Node s = node.next;
        if (s == null || s.isShared())
            unparkSuccessor(node);
    }
}

// 和 release 方法的源碼基本一樣
public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) {
        Node h = head;
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        return true;
    }
    return false;
}

下面來(lái)簡(jiǎn)單說(shuō)明 TestSemaphore 這個(gè)類的邏輯。這個(gè)類持有一個(gè)數(shù)值為 0 的信號(hào)量對(duì)象，并創(chuàng)建了4個(gè)線程，線程 t1 和 t2 用于獲取信號(hào)量，t3 和 t4 則是調(diào)用 release() 方法釋放信號(hào)量。在一般情況下，TestSemaphore 這個(gè)類的代碼都可以正常執(zhí)行。但當(dāng)有極端情況出現(xiàn)時(shí)，可能會(huì)導(dǎo)致同步隊(duì)列掛掉。這里演繹一下這個(gè)極端情況，考慮某次循環(huán)時(shí)，隊(duì)列結(jié)構(gòu)如下：

時(shí)刻1：線程 t3 調(diào)用 unparkSuccessor 方法，head 節(jié)點(diǎn)狀態(tài)由 SIGNAL(-1) 變?yōu)?b>0，并喚醒線程 t1。此時(shí)信號(hào)量數(shù)值為1。

時(shí)刻2：線程 t1 恢復(fù)運(yùn)行，t1 調(diào)用 Semaphore.NonfairSync 的 tryAcquireShared，返回0。然后線程 t1 被切換，暫停運(yùn)行。

時(shí)刻3：線程 t4 調(diào)用 releaseShared 方法，因 head 的狀態(tài)為0，所以 t4 不會(huì)調(diào)用 unparkSuccessor 方法。

時(shí)刻4：線程 t1 恢復(fù)運(yùn)行，t1 成功獲取信號(hào)量，調(diào)用 setHeadAndPropagate。但因?yàn)?propagate = 0，線程 t1 無(wú)法調(diào)用 unparkSuccessor 喚醒線程 t2，t2 面臨無(wú)線程喚醒的情況。因?yàn)?t2 無(wú)法退出等待狀態(tài)，所以 t2.join 會(huì)阻塞主線程，導(dǎo)致程序掛住。

下面再來(lái)看一下修復(fù) BUG 后的代碼，根據(jù) BUG 詳情頁(yè)顯示，該 BUG 在 JDK 1.7 中被修復(fù)。這里找一個(gè) JDK 7 較早版本（JDK 7u10）的代碼看一下，如下：

private void setHeadAndPropagate(Node node, int propagate) {
    Node h = head; // Record old head for check below
    setHead(node);
    
    if (propagate > 0 || h == null || h.waitStatus < 0) {
        Node s = node.next;
        if (s == null || s.isShared())
            doReleaseShared();
    }
}

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

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))
                    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;
    }
}

在按照上面的代碼演繹一下邏輯，如下：

時(shí)刻1：線程 t3 調(diào)用 unparkSuccessor 方法，head 節(jié)點(diǎn)狀態(tài)由 SIGNAL(-1) 變?yōu)?b>0，并喚醒線程t1。此時(shí)信號(hào)量數(shù)值為1。

時(shí)刻2：線程 t1 恢復(fù)運(yùn)行，t1 調(diào)用 Semaphore.NonfairSync 的 tryAcquireShared，返回0。然后線程 t1 被切換，暫停運(yùn)行。

時(shí)刻3：線程 t4 調(diào)用 releaseShared 方法，檢測(cè)到h.waitStatus = 0，t4 將頭節(jié)點(diǎn)等待狀態(tài)由0設(shè)為PROPAGATE(-3)。

時(shí)刻4：線程 t1 恢復(fù)運(yùn)行，t1 成功獲取信號(hào)量，調(diào)用 setHeadAndPropagate。因 propagate = 0，propagate > 0 條件不滿足。而 h.waitStatus = PROPAGATE(-3)，所以條件h.waitStatus < 0成立。進(jìn)而，線程 t1 可以喚醒線程 t2，完成喚醒動(dòng)作的傳播。

到這里關(guān)于狀態(tài) PROPAGATE 的內(nèi)容就講完了。最后，簡(jiǎn)單總結(jié)一下本章開頭提的兩個(gè)問(wèn)題。

問(wèn)題一：PROPAGATE 狀態(tài)用在哪里，以及怎樣向后傳播喚醒動(dòng)作的？
答：PROPAGATE 狀態(tài)用在 setHeadAndPropagate。當(dāng)頭節(jié)點(diǎn)狀態(tài)被設(shè)為 PROPAGATE 后，后繼節(jié)點(diǎn)成為新的頭結(jié)點(diǎn)后。若 propagate > 0 條件不成立，則根據(jù)條件h.waitStatus < 0成立與否，來(lái)決定是否喚醒后繼節(jié)點(diǎn)，即向后傳播喚醒動(dòng)作。

問(wèn)題二：引入 PROPAGATE 狀態(tài)是為了解決什么問(wèn)題？
答：引入 PROPAGATE 狀態(tài)是為了解決并發(fā)釋放信號(hào)量所導(dǎo)致部分請(qǐng)求信號(hào)量的線程無(wú)法被喚醒的問(wèn)題。

聲明：
本章內(nèi)容是在博友 活在夢(mèng)裡 的文章 AbstractQueuedSynchronizer源碼解讀 基礎(chǔ)上，進(jìn)行了一定的補(bǔ)充說(shuō)明。本章所參考的觀點(diǎn)已經(jīng)過(guò)原作者同意，為避免抄襲嫌疑，特此聲明。

到這里，本文就差不多結(jié)束了。如果大家從頭看到尾，到這里就可以放松一下了。寫到這里，我也可以放松一下了。這篇文章總共花費(fèi)了我12天的空閑時(shí)間，確實(shí)不容易。本來(lái)我只打算講一下基本原理，但知道后來(lái)看到本文多次推薦的那篇文章。那篇文章給我的第一感覺(jué)是，作者很厲害。第二感覺(jué)是，我也要寫出一篇較為深入的 AQS 分析文章。雖然寫出來(lái)也不能怎么樣，水平也不會(huì)因此提高多少，也不會(huì)去造個(gè)類似的輪子。但是寫完后，確實(shí)感覺(jué)很有成就感。本文的最后，來(lái)說(shuō)一下如何學(xué)習(xí) AQS 原理。AQS 的大致原理不是很難理解，所以一開始不建議糾結(jié)細(xì)節(jié)，應(yīng)該先弄懂它的大致原理。在此基礎(chǔ)上，再去分析一些細(xì)節(jié)，分析細(xì)節(jié)時(shí)，要從多線程的角度去考慮。比如，有點(diǎn)地方 CAS 失敗后要重試，有的不用重試。總體來(lái)說(shuō) AQS 的大致原理容易理解，細(xì)節(jié)部分比較復(fù)雜。很多細(xì)節(jié)要在腦子里演繹一遍，好好思考才能想通，有點(diǎn)燒腦。另外因?yàn)槲恼缕膯?wèn)題，關(guān)于 AQS ConditionObject 部分的分析將會(huì)放在下一篇文章中進(jìn)行。

最后，再向 AQS 的作者 Doug Lea 致以崇高的敬意。僅盡力弄懂 AQS 的原理都很難了，可想而知，實(shí)現(xiàn) AQS 的難度有多大。

限于本人的能力，加之深入分析 AQS 本身就比較有難度。所以文中難免會(huì)有錯(cuò)誤出現(xiàn)，如果不慎翻車，請(qǐng)見(jiàn)諒。也歡迎在評(píng)論區(qū)指明這些錯(cuò)誤，感謝。

??AbstractQueuedSynchronizer源碼解讀 - 活在夢(mèng)裡

《Java并發(fā)編程的藝術(shù)》 - 方騰飛 / 魏鵬 / 程曉明

本文在知識(shí)共享許可協(xié)議 4.0 下發(fā)布，轉(zhuǎn)載需在明顯位置處注明出處
作者：coolblog
本文同步發(fā)布在我的個(gè)人博客：http://www.coolblog.xyz

本作品采用知識(shí)共享署名-非商業(yè)性使用-禁止演繹 4.0 國(guó)際許可協(xié)議進(jìn)行許可。