一、概述

閑來不卷，隨便聊一點。

(相關(guān)資料圖)

一般情況下，大家系統(tǒng)中至少也是JDK8了，那想必對于JDK5加入的一系列功能并不陌生吧。那時候重點加入了java.util.concurrent并發(fā)包，我們簡稱JUC。JUC下提供了很多并發(fā)編程實用的工具類，比如并發(fā)鎖lock、原子操作atomic、線程池操作Executor等等。下面，我對JUC做了整理，大致分為下面幾點：

基于JDK8，今天重點來聊下JUC并發(fā)包下的一個類，AbstractQueuedSynchronizer。

首先，淺顯的從名字上看，抽象的隊列同步器；實際上，這名字也跟它的作用如出一轍。抽象，即需要被繼承；隊列同步器，其內(nèi)部維護(hù)了一個隊列，供線程入隊等待；最終實現(xiàn)多個線程訪問共享資源的功能。

二、源碼解析

進(jìn)入AbstractQueuedSynchronizer內(nèi)部，需要掌握三個重要的屬性：

private transient volatile Node head;private transient volatile Node tail;private volatile int state;

head：標(biāo)記等待隊列頭部節(jié)點。tail：標(biāo)記等待隊列尾部節(jié)點。state：線程的鎖定狀態(tài)；state=0，表示資源未被上鎖；state>0，表示資源被上鎖

我們調(diào)試AQS的源碼，必須尋找一個源碼調(diào)試的切入點，我這里用我們并發(fā)編程常用的Lock鎖作為調(diào)試AQS的切入點，因為這是解決線程安全問題常用的手段之一。

2.1、源碼的切入點

AQS的源碼調(diào)試，從Lock接口出發(fā)，JDK源碼定義如下：

public interface Lock {    void lock();    void lockInterruptibly() throws InterruptedException;    boolean tryLock();    boolean tryLock(long time, TimeUnit unit) throws InterruptedException;    void unlock();    Condition newCondition();}

從源碼中看到，Lock?是一個接口，所以該接口會有一些實現(xiàn)類，其中有一個實現(xiàn)類ReentrantLock，可重入鎖，想必大家都不會陌生。

2.2、ReentrantLock的lock方法

通過跟蹤源碼可以看到，ReentrantLock#lock內(nèi)部實現(xiàn)貌似比較簡單，只有簡短的一行代碼

public void lock() {    sync.lock();}

其實內(nèi)部是維護(hù)了一個Sync?的抽象類，調(diào)用的是Sync的lock()方法。

abstract static class Sync extends AbstractQueuedSynchronizer {    private static final long serialVersionUID = -5179523762034025860L;    abstract void lock();    final boolean nonfairTryAcquire(int acquires) {        final Thread current = Thread.currentThread();        int c = getState();        if (c == 0) {            if (compareAndSetState(0, acquires)) {                setExclusiveOwnerThread(current);                return true;            }        }        else if (current == getExclusiveOwnerThread()) {            int nextc = c + acquires;            if (nextc < 0) // overflow                throw new Error("Maximum lock count exceeded");            setState(nextc);            return true;        }        return false;    }    protected final boolean tryRelease(int releases) {        int c = getState() - releases;        if (Thread.currentThread() != getExclusiveOwnerThread())            throw new IllegalMonitorStateException();        boolean free = false;        if (c == 0) {            free = true;            setExclusiveOwnerThread(null);        }        setState(c);        return free;    }    protected final boolean isHeldExclusively() {        return getExclusiveOwnerThread() == Thread.currentThread();    }    // ...}

可以看到，Sync?也是個抽象類，它有兩個實現(xiàn)類：NonfairSync和FairSync?，這里其實就引出了我們今天的主角，AbstractQueuedSynchronizer，Sync繼承了它。

static final class NonfairSync extends Sync {    private static final long serialVersionUID = 7316153563782823691L;    final void lock() {        if (compareAndSetState(0, 1))            setExclusiveOwnerThread(Thread.currentThread());        else            acquire(1);    }    protected final boolean tryAcquire(int acquires) {        return nonfairTryAcquire(acquires);    }}

static final class FairSync extends Sync {    private static final long serialVersionUID = -3000897897090466540L;    final void lock() {        acquire(1);    }    protected final boolean tryAcquire(int acquires) {        final Thread current = Thread.currentThread();        int c = getState();        if (c == 0) {            if (!hasQueuedPredecessors() &&                compareAndSetState(0, acquires)) {                setExclusiveOwnerThread(current);                return true;            }        }        else if (current == getExclusiveOwnerThread()) {            int nextc = c + acquires;            if (nextc < 0)                throw new Error("Maximum lock count exceeded");            setState(nextc);            return true;        }        return false;    }}

下面我整理了這一系列類的UML圖

通過類圖可知，lock()方法最終調(diào)用的是ReentrantLock?類下，內(nèi)部類NonfairSync或FairSync?的lock方法；對于這兩個類，前者叫非公平鎖，后者叫公平鎖。通過ReentrantLock?的構(gòu)造器可知，默認(rèn)使用NonfairSync類。

public ReentrantLock() {    sync = new NonfairSync();}

從NonfairSync類的lock方法出發(fā)，引出第一個AQS下的方法compareAndSetState。

final void lock() {    if (compareAndSetState(0, 1))        setExclusiveOwnerThread(Thread.currentThread());    else        acquire(1);}

從compareAndSetState方法的命名可以發(fā)現(xiàn)，就是比較并交換的意思，典型的CAS無鎖機(jī)制。

protected final boolean compareAndSetState(int expect, int update) {    // See below for intrinsics setup to support this    return unsafe.compareAndSwapInt(this, stateOffset, expect, update);}

我們可以觀察到，這里其實調(diào)用的是Unsafe類的compareAndSwapInt方法，傳入的expect為0，update為1；意思是如果當(dāng)前值為0，那我就把值最終更新為1。

Unsafe?這個類下面，發(fā)現(xiàn)好多方法都是用native?這個關(guān)鍵詞進(jìn)行修飾的（也包括compareAndSwapInt方法），用native關(guān)鍵詞修飾的方法，表示原生的方法；原生方法的實現(xiàn)并不是Java語言，最終實現(xiàn)是C/C++；這并不是本文的討論范圍。

回到AQS的compareAndSetState方法，返回值是boolean類型，true表示值更新為1成功，false表示不成功。這里出現(xiàn)兩個分支，成功，走setExclusiveOwnerThread方法；不成功，走acquire方法。咱優(yōu)先討論acquire方法。

2.3、AQS的acquire方法

先來看一下該方法的源碼；

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

這里的核心是兩個方法，tryAcquire方法和acquireQueued方法。首先會調(diào)用tryAcquire()方法，看方法命名是嘗試獲取；實際上這個方法確實就在做一件事“嘗試獲取資源”。

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

不過AQS中的這個方法是protected?修飾，并沒有去實現(xiàn)，僅僅只是預(yù)留了方法入口，后期需要由其子類去實現(xiàn)；這里的子類就是上文中的NonfairSync類，該類的源碼在上文中已經(jīng)貼出。這段源碼其實運(yùn)用了我們常見的一個設(shè)計模式，“模板方法模式”。

2.4、NonfairSync的tryAcquire方法

NonfairSync的tryAcquire方法源碼如下

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

這里并沒有直接去實現(xiàn)tryAcquire方法，而是調(diào)用了Sync類下的nonfairTryAcquire方法。

final boolean nonfairTryAcquire(int acquires) {    final Thread current = Thread.currentThread();    int c = getState();    if (c == 0) {        if (compareAndSetState(0, acquires)) {            setExclusiveOwnerThread(current);            return true;        }    }    else if (current == getExclusiveOwnerThread()) {        int nextc = c + acquires;        if (nextc < 0) // overflow            throw new Error("Maximum lock count exceeded");        setState(nextc);        return true;    }    return false;}

這里有個getState方法，最終返回的是AQS中的state字段，這個字段就是多個線程搶占的共享資源，所以這個字段很重要；volatile?關(guān)鍵字修飾，保證內(nèi)存的可見性，int類型，對于ReentrantLock鎖而言，當(dāng)state=0時，表示無鎖，當(dāng)state>0時，表示資源已被線程鎖定。

下面分析下這段代碼：

如果state=0表示無鎖，通過cas去更新state的值，這里更新為1。將持有鎖的線程更新為當(dāng)前線程。如果上述cas未更新成功，或者state!=0，表示已上鎖。繼續(xù)判斷下持有鎖的線程如果是當(dāng)前線程，state字段做疊加，這里表示ReentrantLock的含義，表示可重入鎖。最后，state!=0，持有鎖的線程也不是當(dāng)前線程，表示不能對資源加鎖，返回false。

tryAcquire方法的判斷至此結(jié)束，不過最終的走向需要看它的返回值；返回true，表示當(dāng)前線程搶占到鎖，或者當(dāng)前線程就是搶占鎖的線程，直接重入，加鎖流程結(jié)束；返回false，表示沒有搶占到鎖，流程繼續(xù)，這里就引出下個話題，CLH線程等待隊列。

2.5、AQS的addWaiter方法2.5.1、CLH隊列

首先咱來看一段源碼中的注釋

The wait queue is a variant of a "CLH" (Craig, Landin, and Hagersten) lock queue. CLH locks are normally used for spinlocks

大致意思是：CLH隊列是由Craig、Landin、Hagersten這三位老哥名字的首字母疊加在一起命名的，它是一個等待隊列，它是一個變種隊列，用到了自旋。

這里的信息要抓住三點：等待隊列、變種隊列、自旋。

2.5.2、Node類

在解析addWaiter方法實現(xiàn)之前，就不得不提到一個內(nèi)部類Node；addWaiter方法的入?yún)⑹沁@個類型，所以先來看看這個類。源碼如下：

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;    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) {     // Used by addWaiter        this.nextWaiter = mode;        this.thread = thread;    }    Node(Thread thread, int waitStatus) { // Used by Condition        this.waitStatus = waitStatus;        this.thread = thread;    }}

這里先大致介紹下，每個屬性的意思：

SHARED：類型就是Node，表示共享模式。EXCLUSIVE：類型也是Node，表示獨占模式，這里的ReentrantLock就是獨占模式。waitStatus：int類型，當(dāng)前Node節(jié)點下，存儲的線程狀態(tài)。CANCELLED：int類型，等于1，waitStatus屬性的值之一，表示節(jié)點被取消狀態(tài)。SIGNAL：int類型，等于-1，waitStatus屬性的值之一，表示當(dāng)前節(jié)點需要去喚醒下一個節(jié)點。CONDITION：int類型，等于-2，waitStatus屬性的值之一，表示節(jié)點處于等待狀態(tài)。PROPAGATE：int類型，等于-2，waitStatus屬性的值之一，表示下一個被獲取的對象應(yīng)該要無條件傳播，該值僅在共享模式下使用。prev：Node類型，指向隊列中當(dāng)前節(jié)點的前一個節(jié)點。next：Node類型，指向隊列中當(dāng)前節(jié)點的下一個節(jié)點。thread：存儲當(dāng)前線程信息。nextWaiter：用來存儲節(jié)點的指針，不過會出現(xiàn)兩種情況；等待隊列中，會將該屬性的值設(shè)置成SHARED或者EXCLUSIVE，用來區(qū)分當(dāng)前節(jié)點處于共享模式還是獨享模式；條件隊列中，用于存放下一個節(jié)點的指針，所以當(dāng)是條件隊列的情況下，這個隊列是單向隊列。isShared()：返回是否屬于共享模式，true表示共享模式，false表示獨享模式。predecessor()：獲取當(dāng)前節(jié)點的前一個節(jié)點。

另外，Node類還有兩個有參構(gòu)造器：從作者的注釋就能看出來，第一個構(gòu)造器是在等待隊列的時，創(chuàng)建節(jié)點使用，第二個構(gòu)造器是在條件隊列時，創(chuàng)建節(jié)點使用。

2.5.3、方法解析

private Node addWaiter(Node mode) {    Node node = new Node(Thread.currentThread(), mode);    Node pred = tail;    if (pred != null) {        node.prev = pred;        if (compareAndSetTail(pred, node)) {            pred.next = node;            return node;        }    }    enq(node);    return node;}private Node enq(final Node node) {    for (;;) {        Node t = tail;        if (t == null) { // Must initialize            if (compareAndSetHead(new Node()))                tail = head;        } else {            node.prev = t;            if (compareAndSetTail(t, node)) {                t.next = node;                return t;            }        }    }}

其實這段方法是在創(chuàng)建Node對象，Node對象就是組成CLH隊列的基礎(chǔ)元素。

創(chuàng)建一個Node對象，mode參數(shù)由上述的acquire()方法傳遞而來，可以看到傳入Node.EXCLUSIVE，表示獨占模式。判斷隊尾有指向節(jié)點，剛創(chuàng)建的節(jié)點放入隊列的隊尾，并且通過cas將隊尾指針改成當(dāng)前創(chuàng)建節(jié)點，最后返回當(dāng)前創(chuàng)建節(jié)點。如果隊尾沒有指向節(jié)點，調(diào)用enq方法，做隊列的初始化操作。這里出現(xiàn)了第一個自旋，enq方法是無限循環(huán)的，就像作者注釋的一樣，Must initialize，必須初始化。這里先是重新new了一個新的node（也可以叫空節(jié)點），標(biāo)記它為隊列頭。隨后再將addWaiter方法中創(chuàng)建的node，加入到隊列尾。

總結(jié)下addWaiter方法干的事情：

創(chuàng)建一個節(jié)點，存儲當(dāng)前線程，并標(biāo)記獨占模式。判斷隊列是否為空，不為空，通過cas將存儲當(dāng)前線程的node節(jié)點加入到對尾，并且對該節(jié)點做對尾標(biāo)記。隊列為空，通過自旋，做初始化操作。初始化過后的隊列，隊列頭是一個空節(jié)點，隊列尾是存儲當(dāng)前線程的節(jié)點。2.6、AQS的acquireQueued方法

還是先來看下這個方法的源碼；

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)) {                setHead(node);                p.next = null; // help GC                failed = false;                return interrupted;            }            if (shouldParkAfterFailedAcquire(p, node) &&                parkAndCheckInterrupt())                interrupted = true;        }    } finally {        if (failed)            cancelAcquire(node);    }}private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {    int ws = pred.waitStatus;    if (ws == Node.SIGNAL)        return true;    if (ws > 0) {        do {            node.prev = pred = pred.prev;        } while (pred.waitStatus > 0);        pred.next = node;    } else {        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);    }    return false;}private void cancelAcquire(Node node) {    if (node == null)        return;    node.thread = null;    Node pred = node.prev;    while (pred.waitStatus > 0)        node.prev = pred = pred.prev;    Node predNext = pred.next;    node.waitStatus = Node.CANCELLED;    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)                compareAndSetNext(pred, predNext, next);        } else {            unparkSuccessor(node);        }        node.next = node; // help GC    }}

從這個方法看到，又是運(yùn)用了無限循環(huán)，需要分兩個步驟去觀察：1.當(dāng)前方法中的判斷，自己的上一個節(jié)點是否是頭部節(jié)點（頭部節(jié)點就是占用資源的節(jié)點）；2.當(dāng)前節(jié)點正式入隊列，并且被掛起。

2.6.1、acquireQueued方法中的判斷

當(dāng)前節(jié)點的前一個節(jié)點是隊列頭部，意味著當(dāng)前節(jié)點的前一個節(jié)點，就是持有資源的節(jié)點；當(dāng)資源被釋放，當(dāng)前節(jié)點會去嘗試爭奪鎖資源；如果拿到鎖資源，當(dāng)前節(jié)點會被標(biāo)記為隊列頭部節(jié)點，它的上個節(jié)點（老的頭部節(jié)點）會被置為null，需要被GC及時清除，所以作者在這里添加了一個注釋：help GC；下圖就是描述了這個流程：

2.6.2、shouldParkAfterFailedAcquire方法實現(xiàn)

如果當(dāng)前節(jié)點的上一個節(jié)點，并不是頭部節(jié)點；這里就需要用到上述Node類中介紹的各種狀態(tài)字段了；先來重點介紹下Node類中的兩個狀態(tài)屬性：

CANCELLED：int類型，等于1，waitStatus屬性的值之一，表示節(jié)點被取消。SIGNAL：int類型，等于-1，waitStatus屬性的值之一，表示當(dāng)前節(jié)點需要去喚醒下一個節(jié)點。

進(jìn)入的shouldParkAfterFailedAcquire這個方法內(nèi)部，該方法接受兩個參數(shù)：當(dāng)前節(jié)點前一個節(jié)點和當(dāng)前節(jié)點。首先，獲取上一個節(jié)點的waitStatus屬性，然后通過這個屬性做如下判斷：

如果狀態(tài)是SIGNAL（即等于-1），直接返回true，后續(xù)就會交給parkAndCheckInterrupt方法去將當(dāng)前線程掛起。如果不是SIGNAL，對于當(dāng)前ReentrantLock而言，ws>0的操作是滿足的，所以下面的步驟就是當(dāng)前節(jié)點一直往前尋找，跳過已被標(biāo)記狀態(tài)為CANCELLED的節(jié)點，直到找到狀態(tài)是SIGNAL的節(jié)點，將該節(jié)點作為當(dāng)前節(jié)點的上一個節(jié)點。也印證了SIGNAL狀態(tài)的解釋：當(dāng)前節(jié)點的上一個節(jié)點是SIGNAL，那么當(dāng)前節(jié)點需要掛起，等待被喚醒。最后進(jìn)入下個循環(huán)，直到上個節(jié)點狀態(tài)是SIGNAL，執(zhí)行上面的第一步，返回true。

這里可以想象成一個排隊去食堂打飯的場景，你在低頭玩手機(jī)前，跟你前面的同學(xué)說，我玩會手機(jī)，快到了叫我一下；結(jié)果你前面的同學(xué)嫌隊伍長走了（CANCELLED狀態(tài)），所以你只能繼續(xù)找他的上一個同學(xué)；直到有同學(xué)回答你，好的（該同學(xué)被標(biāo)記SIGNAL狀態(tài)）；然后你就低頭玩手機(jī)，等待回答你“好的”的那個同學(xué)叫你。

最后compareAndSetWaitStatus方法其實不用看也知道，通過cas機(jī)制，將當(dāng)前節(jié)點的上一個節(jié)點的waitStatus修改成SIGNAL狀態(tài)，這樣的話，當(dāng)前節(jié)點才能被掛起，等待喚醒。

再來看下parkAndCheckInterrupt這個方法

private final boolean parkAndCheckInterrupt() {    LockSupport.park(this);    return Thread.interrupted();}// LockSupport#parkpublic static void park(Object blocker) {    Thread t = Thread.currentThread();    setBlocker(t, blocker);    UNSAFE.park(false, 0L);    setBlocker(t, null);}

其中最終又是這個Unsafe類，通過它的原生方法park，去掛起當(dāng)前線程，這里就不展開贅述了。

2.7、資源上鎖總結(jié)

下面整理下從lock方法作為切入點，一系列的調(diào)用：

2.8、ReentrantLock的unlock方法

之前一直在講資源“上鎖”，那么這個方法就是給資源解鎖。這里給出重要的部分源碼

// AQS中public final boolean release(int arg) {    if (tryRelease(arg)) {        Node h = head;        if (h != null && h.waitStatus != 0)            unparkSuccessor(h);        return true;    }    return false;}// AQS中private void unparkSuccessor(Node node) {    int ws = node.waitStatus;    if (ws < 0)        compareAndSetWaitStatus(node, ws, 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)        LockSupport.unpark(s.thread);}// ReentrantLock中protected final boolean tryRelease(int releases) {    int c = getState() - releases;    if (Thread.currentThread() != getExclusiveOwnerThread())        throw new IllegalMonitorStateException();    boolean free = false;    if (c == 0) {        free = true;        setExclusiveOwnerThread(null);    }    setState(c);    return free;}

2.9、ReentrantLock的tryRelease方法

在調(diào)用unlock方法去解鎖后，最終是調(diào)用AQS中的release方法去實現(xiàn)這個解鎖功能的；在該方法中，首先會調(diào)用ReentrantLock中的tryRelease方法，去做state狀態(tài)值的遞減操作。

首先，獲取state值（在AQS中有這個公共屬性，上文提到過），這里是對當(dāng)前state值減去1。再判斷當(dāng)前解鎖的線程與持有鎖的線程是不是同一個，不是的話，直接拋異常。所以t線程占用鎖，只有t線程才能解鎖，解鈴還須系鈴人。最后判斷做完遞減的值是不是等于0，如果為0，將持有鎖的線程清空，更新state字段為遞減值（這里是0），最后返回true，代表鎖已經(jīng)被釋放了。如果不是0，更新state字段為遞減值（不是0），也不會清空持有鎖的線程，意味著資源還是被線程加鎖中，最后返回false。2.10、AQS的release方法

在tryRelease方法返回false的時候，release方法并不會做任何操作，直接就結(jié)束了，意味著解鎖并沒有完成；但是在返回true的時候，具體分以下幾部操作：

拿到CLH隊列被標(biāo)記頭部的節(jié)點。判斷不是空（隊列不能是空的），并且頭部節(jié)點的等待狀態(tài)不是0，在這種情況下，它只能是-1（SIGNAL），所以是需要去喚醒下個節(jié)點的。最后，調(diào)用AQS中的unparkSuccessor方法，去喚醒線程。2.11、AQS的unparkSuccessor方法

上面說到了，這個方法主要是用來喚醒線程的，下面還是做一下具體的解析：

該方法傳參是一個Node節(jié)點，這里傳入的是被標(biāo)記隊列頭的節(jié)點（頭部節(jié)點是持有鎖資源的節(jié)點）。拿到頭部節(jié)點的waitStatus狀態(tài)屬性，并且判斷小于0的情況下（該情況是waitStatus=-1），通過cas機(jī)制將頭部節(jié)點的狀態(tài)改為0，初始化狀態(tài)。拿到頭部節(jié)點的下個節(jié)點，也就是真正意義上處于等待中的第一個節(jié)點。它還是先判斷了這個拿到的節(jié)點是否為null，或者狀態(tài)大于0（亦或說判斷狀態(tài)等于1）；如果條件成立，說明頭節(jié)點的下個節(jié)點是空，或者下個節(jié)點被取消了。如果第四個判斷條件滿足，從隊尾一直從后往前找，找到離頭節(jié)點最近的那個節(jié)點。通過Unsafe類的unpark原生方法去喚醒上述找到的，距離頭部節(jié)點最近的未處于取消狀態(tài)下的節(jié)點。2.12、資源解鎖總結(jié)

通過上面的描述可以發(fā)現(xiàn)，資源解鎖是相對簡單的；它只能被上鎖的線程去解鎖；通過遞減AQS內(nèi)部維護(hù)的state屬性值，直到state減為0，表示資源已被解鎖；當(dāng)資源被解鎖后，需要通過Unsafe的unpark方法，去喚醒CLH隊列中，被掛起的第一個節(jié)點上的線程。

2.13、公平鎖與非公平鎖的差異

在2.2中說過，當(dāng)我們使用無參構(gòu)造器創(chuàng)建一把“鎖”的時候，默認(rèn)是使用NonfairSync這個內(nèi)部類，也就是非公平鎖；但是在源碼中發(fā)現(xiàn)ReentrantLock?還存在一個有參構(gòu)造器，參數(shù)是一個boolean類型；

public ReentrantLock(boolean fair) {    sync = fair ? new FairSync() : new NonfairSync();}

很明顯，這種方式就是將選擇權(quán)交給開發(fā)人員，當(dāng)我們傳入true時，就會創(chuàng)建一把“公平鎖”。還是一樣，先來看下公平鎖的內(nèi)部；

static final class FairSync extends Sync {    private static final long serialVersionUID = -3000897897090466540L;    final void lock() {        acquire(1);    }    protected final boolean tryAcquire(int acquires) {        final Thread current = Thread.currentThread();        int c = getState();        if (c == 0) {            if (!hasQueuedPredecessors() &&                compareAndSetState(0, acquires)) {                setExclusiveOwnerThread(current);                return true;            }        }        else if (current == getExclusiveOwnerThread()) {            int nextc = c + acquires;            if (nextc < 0)                throw new Error("Maximum lock count exceeded");            setState(nextc);            return true;        }        return false;    }}

從源碼的角度，咱來看下，為什么一個叫“非公平鎖”，另一個叫“公平鎖”？

其實不難發(fā)現(xiàn)，NonfairSync?內(nèi)部的lock方法，它是一上來就通過cas機(jī)制去搶占state公共資源，搶不到才去執(zhí)行acquire方法實現(xiàn)后續(xù)入隊列等一系列的操作；而這里FairSync的lock方法，它是直接執(zhí)行acquire方法，執(zhí)行后續(xù)的操作。等于非公平鎖，會去多爭取一次資源，對于在CLH隊列中等待的線程，是“不公平”的。

除了lock方法存在差異之外，在tryAcquire方法中，也存在著不同。FairSync類中，會多執(zhí)行hasQueuedPredecessors方法，它是AQS下的一個公用方法，下面具體看下這個方法；

public final boolean hasQueuedPredecessors() {    Node t = tail;    Node h = head;    Node s;    return h != t &&        ((s = h.next) == null || s.thread != Thread.currentThread());}

只有簡短的幾行，卻有很多種可能性，但是整個方法主要功能就是判斷當(dāng)前線程是否需要入隊列：返回false，隊列為空，不對等待；返回true，隊列不是空，去排隊等待。下面需要重點講下這一行代碼：return h != t && ((s = h.next) == null || s.thread != Thread.currentThread());

2.13.1、hasQueuedPredecessors返回false

返回false，情況也有兩種：1、h != t** **是false，2、h != t是true，并且 (s = h.next) == null 是false， s.thread != Thread.currentThread()是false。

第一種情況比較簡單，意思是頭結(jié)點和尾節(jié)點是同一個，說明隊列是空的，不需要排隊等待，所以直接返回false。

第二種情況，頭尾不是同一個節(jié)點，頭部節(jié)點的下個節(jié)點也不是空，并且頭部節(jié)點的下一個節(jié)點就是當(dāng)前線程。其實就可以理解為，前面的資源剛釋放，正好輪到當(dāng)前線程來搶占資源，這種情況相對較少。

2.13.2、hasQueuedPredecessors返回true

返回true，有兩種情況：

h != t是true，并且 (s = h.next) == null 是true。

h != t是true，并且 (s = h.next) == null 是false， s.thread不等于Thread.currentThread()是true。

1、這里的頭尾不是同一個節(jié)點是必要滿足的條件，保證了隊列起碼不是空的。然后(s = h.next) == null 滿足是true，這里解釋起來就必須回顧下enq初始化隊列這個方法。

private Node enq(final Node node) {    for (;;) {        Node t = tail;        if (t == null) { // Must initialize            if (compareAndSetHead(new Node()))                tail = head;        } else {            node.prev = t;            if (compareAndSetTail(t, node)) {                t.next = node;                return t;            }        }    }}

從這個方法可知，先是將節(jié)點的prev指向前一個節(jié)點，然后再通過cas修改尾部標(biāo)識，最后再將前一個節(jié)點的next指向當(dāng)前節(jié)點；因此AQS，入隊操作是非原子性的。

繼續(xù)回到判斷本身，頭部節(jié)點拿到鎖在執(zhí)行；中間節(jié)點沒拿到鎖在入隊；此時頭部節(jié)點執(zhí)行完后釋放鎖，當(dāng)前節(jié)點嘗試不入隊拿鎖，但是中間線程已經(jīng)在排隊了，但是還沒來得及執(zhí)行t.next = node的操作，導(dǎo)致(s = h.next) == null 滿足，所以當(dāng)前節(jié)點必須入隊，最終返回true。

2、滿足s.thread != Thread.currentThread()的情況，執(zhí)行到這里，可以明確隊列首先不是空，并且h.next != null，也就是頭節(jié)點之后還有其他節(jié)點，最后再判斷了下，s.thread != Thread.currentThread為true，也就是頭節(jié)點的下個節(jié)點并不是當(dāng)前節(jié)點，既然如此，那只能乖乖去隊列中排隊了，所以最終返回true。

三、業(yè)務(wù)運(yùn)用

想必大家對于并發(fā)鎖并不陌生了，上文我也是通過ReentrantLock?這個并發(fā)鎖為入口，一步步來解析AQS中的實現(xiàn)。所以這里就不用ReentrantLock舉例，這里換一個同步工具：CountDownLatch，它也是基于AQS來實現(xiàn)的。

CountDownLatch是通過一個計數(shù)器來實現(xiàn)的，初始值為線程的數(shù)量。每當(dāng)一個線程完成了自己的任務(wù)，計數(shù)器的值就相應(yīng)得減1。當(dāng)計數(shù)器到達(dá)0時，表示所有的線程都已執(zhí)行完畢,然后在等待的線程就可以恢復(fù)執(zhí)行任務(wù)。

這個其實跟ReentrantLock思路差不多，一個是state初始值就是0，通過“上鎖”一步步疊加這個值；一個是state讓使用者自己設(shè)定初始值，通過線程調(diào)用，一步步遞減這個值。

CountDownLatch具體的運(yùn)用情況如下：1、一個主線程中，需要開啟多個子線程，并且要在多個子線程執(zhí)行完畢后，主線程才能繼續(xù)往下執(zhí)行。2、通過多個線程一起執(zhí)行，提高執(zhí)行的效率。

下面，通過一個真實的業(yè)務(wù)場景，來進(jìn)一步了解下CountDownLatch這個同步工具，具體是怎么使用的。

現(xiàn)在有這么一個接口，查詢用戶的詳情信息；用戶信息由五部分組成：1、用戶基本信息；2、用戶影像信息；3、用戶工商信息；4、用戶賬戶信息；5、用戶組織架構(gòu)信息；按照原本的邏輯是按照順序1-5這樣一步步查詢，最后組裝用戶VO對象，接口返回。但是這里可以用上CountDownLatch這個工具類，申請五個線程，分別去查詢這五種信息，提高接口效率。

/** * @author 往事如風(fēng) * @version 1.0 * @date 2023/4/11 18:10 * @description：數(shù)據(jù)查詢 */@RestControllerpublic class QueryController {    @GetMapping("/query")    public Result download() throws InterruptedException {        ExecutorService executorService = Executors.newFixedThreadPool(5);        // 模擬查詢數(shù)據(jù)        List row1 = CollUtil.newArrayList("aa", "bb", "cc", "dd");        List row2 = CollUtil.newArrayList("aa1", "bb1", "cc1", "dd1");        List row3 = CollUtil.newArrayList("aa2", "bb2", "cc2", "dd2");        List row4 = CollUtil.newArrayList("aa3", "bb3", "cc3", "dd3");        List row5 = CollUtil.newArrayList("aa4", "bb4", "cc4", "dd4");        CountDownLatch count = new CountDownLatch(5);        DataQuery d = new DataQuery();        // 開始時間        long start = System.currentTimeMillis();        System.out.println("開始查詢數(shù)據(jù)。。。。");        executorService.execute(() -> {            System.out.println("查詢用戶基本信息。。。。。。");            try {                Thread.sleep(1000);            } catch (InterruptedException e) {                throw new RuntimeException(e);            }            d.setBaseInfo(row1);            count.countDown();        });        executorService.execute(() -> {            System.out.println("查詢用戶影像信息。。。。。。");            try {                Thread.sleep(1000);            } catch (InterruptedException e) {                throw new RuntimeException(e);            }            d.setImgInfo(row2);            count.countDown();        });        executorService.execute(() -> {            System.out.println("查詢用戶工商信息。。。。。。");            try {                Thread.sleep(1000);            } catch (InterruptedException e) {                throw new RuntimeException(e);            }            d.setBusinessInfo(row3);            count.countDown();        });        executorService.execute(() -> {            System.out.println("查詢用戶賬戶信息。。。。。。");            try {                Thread.sleep(1000);            } catch (InterruptedException e) {                throw new RuntimeException(e);            }            d.setAccountInfo(row4);            count.countDown();        });        executorService.execute(() -> {            System.out.println("查詢用戶組織架構(gòu)信息。。。。。。");            try {                Thread.sleep(1000);            } catch (InterruptedException e) {                throw new RuntimeException(e);            }            d.setOrgInfo(row5);            count.countDown();        });        // 阻塞：直到count的值減為0        count.await();        executorService.shutdown();        // 結(jié)束時間        long end = System.currentTimeMillis();        System.out.println("查詢結(jié)束。。。。。");        System.out.println("用時時間：" + (end - start));        return Result.success(d);    }    @Data    class DataQuery {        private List baseInfo;        private List imgInfo;        private List businessInfo;        private List accountInfo;        private List orgInfo;    }}/*控制臺輸出：開始查詢數(shù)據(jù)。。。。查詢用戶基本信息。。。。。。查詢用戶影像信息。。。。。。查詢用戶工商信息。。。。。。查詢用戶賬戶信息。。。。。。查詢用戶組織架構(gòu)信息。。。。。。查詢結(jié)束。。。。。用時時間：1017*/

這段代碼做了模擬查詢各種用戶信息的操作，其中每個線程都暫停1秒，代表在查詢這五種數(shù)據(jù)；最終打印的用時時間是1017ms，說明這五個線程是同時進(jìn)行的，大大提高了接口的效率。

四、寫在最后

AQS提供了一個FIFO隊列，這里稱為CLH隊列，可以看成是一個用來實現(xiàn)同步鎖以及其他涉及到同步功能的核心組件，常見的有:ReentrantLock、CountDownLatch、Semaphore等。

AQS是一個抽象類，主要是通過繼承的方式來使用，它本身沒有實現(xiàn)任何的同步接口，僅僅是定義了同步狀態(tài)的獲取以及釋放的方法來提供自定義的同步組件。

可以這么說，只要搞懂了AQS，那么J.U.C中絕大部分的api都能輕松掌握。

本文主要提供了從ReentrantLock?出發(fā)，解析了AQS中的各種公用的方法，如果需要知道其他類中怎么去使用AQS中的方法，其實也只需要找到切入點，一步步調(diào)試下去即可，不過，我想很多地方都是和ReentrantLock中一致的。

五、參考源碼

編程文檔：https://gitee.com/cicadasmile/butte-java-note應(yīng)用倉庫：https://gitee.com/cicadasmile/butte-flyer-parent

關(guān)鍵詞：