前言：Java并發(fā)包很多的同步工具類底層都是基于AQS來實(shí)現(xiàn)的，比如我們工作中經(jīng)常用的Lock工具ReentrantLock、柵欄CountDownLatch、信號(hào)量Semaphore等。如果你想深入研究Java并發(fā)編程的話，那么AQS一定是繞不開的一塊知識(shí)點(diǎn)，而且關(guān)于AQS的知識(shí)點(diǎn)也是面試中經(jīng)?？疾斓膬?nèi)容，所以深入學(xué)習(xí)AQS很有必要。 

學(xué)習(xí)AQS之前，我們有必要了解一下AQS底層中大量使用的CAS：Java多線程10：并發(fā)編程的基石CAS機(jī)制

一、AQS簡(jiǎn)介

1.1、AOS概念

AQS，全名AbstractQueuedSynchronizer，是一個(gè)抽象類的隊(duì)列式同步器，它的內(nèi)部通過維護(hù)一個(gè)狀態(tài)volatile int state(共享資源)，一個(gè)FIFO線程等待隊(duì)列來實(shí)現(xiàn)同步功能。AQS類是整個(gè) JUC包的核心類，JUC 中的ReentrantLock、ReentrantReadWriteLock、CountDownLatch、Semaphore和LimitLatch等同步工具都是基于AQS實(shí)現(xiàn)的。

state用關(guān)鍵字volatile修飾，代表著該共享資源的狀態(tài)一更改就能被所有線程可見，而AQS的加鎖方式本質(zhì)上就是多個(gè)線程在競(jìng)爭(zhēng)state，當(dāng)state為0時(shí)代表線程可以競(jìng)爭(zhēng)鎖，不為0時(shí)代表當(dāng)前對(duì)象鎖已經(jīng)被占有，其他線程來加鎖時(shí)則會(huì)失敗，加鎖失敗的線程會(huì)被放入一個(gè)FIFO的等待隊(duì)列中，這些線程會(huì)被UNSAFE.park()操作掛起，等待其他獲取鎖的線程釋放鎖才能夠被喚醒。

而這個(gè)等待隊(duì)列其實(shí)就相當(dāng)于一個(gè)CLH隊(duì)列，用一張?jiān)韴D來表示大致如下：

 1.2、AQS的核心思想

如果被請(qǐng)求的共享資源空閑，則將當(dāng)前請(qǐng)求資源的線程設(shè)置為有效的工作線程，并將共享資源設(shè)置為鎖定狀態(tài)，如果被請(qǐng)求的共享資源被占用，那么就需要一套線程阻塞等待以及被喚醒時(shí)鎖分配的機(jī)制，這個(gè)機(jī)制AQS是用CLH隊(duì)列鎖實(shí)現(xiàn)的，即將暫時(shí)獲取不到鎖的線程加入到隊(duì)列中。CLH（Craig，Landin，and Hagersten）隊(duì)列是一個(gè)虛擬的雙向隊(duì)列，虛擬的雙向隊(duì)列即不存在隊(duì)列實(shí)例，僅存在節(jié)點(diǎn)之間的關(guān)聯(lián)關(guān)系。
AQS是將每一條請(qǐng)求共享資源的線程封裝成一個(gè)CLH鎖隊(duì)列的一個(gè)結(jié)點(diǎn)（Node），來實(shí)現(xiàn)鎖的分配。

用大白話來說，AQS就是基于CLH隊(duì)列，用volatile修飾共享變量state，線程通過CAS去改變狀態(tài)符，成功則獲取鎖成功，失敗則進(jìn)入等待隊(duì)列，等待被喚醒。

1.3、AQS是自旋鎖

AQS是自旋鎖：在等待喚醒的時(shí)候，經(jīng)常會(huì)使用自旋（while(!cas())）的方式，不停地嘗試獲取鎖，直到被其他線程獲取成功

實(shí)現(xiàn)了AQS的鎖有：自旋鎖、互斥鎖、讀鎖寫鎖、條件產(chǎn)量、信號(hào)量、柵欄都是AQS的衍生物

1.4、AQS支持兩種資源分享的方式 

Exclusive（獨(dú)占，只有一個(gè)線程能執(zhí)行，如ReentrantLock）和Share（共享，多個(gè)線程可同時(shí)執(zhí)行，如Semaphore/CountDownLatch）。

自定義的同步器繼承AQS后，只需要實(shí)現(xiàn)共享資源state的獲取和釋放方式即可，其他如線程隊(duì)列的維護(hù)（如獲取資源失敗入隊(duì)/喚醒出隊(duì)等）等操作，AQS在底層已經(jīng)實(shí)現(xiàn)了。

線程的阻塞和喚醒

在JDK1.5之前，除了內(nèi)置的監(jiān)視器機(jī)制外，沒有其它方法可以安全且便捷得阻塞和喚醒當(dāng)前線程。

JDK1.5以后，java.util.concurrent.locks包提供了LockSupport類來作為線程阻塞和喚醒的工具。

二、AQS原理

2.1、同步狀態(tài)的管理

同步狀態(tài)，其實(shí)就是資源。AQS使用單個(gè)int（32位）來保存同步狀態(tài)，并暴露出getState、setState以及compareAndSetState操作來讀取和更新這個(gè)狀態(tài)。

public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer
    implements java.io.Serializable {
 
    private volatile int state;
  
    protected final int getState() {
         return state;
    }
 
    protected final void setState(int newState) {
       state = newState;
    }
 
    protected final boolean compareAndSetState(int expect, int update) {
     // See below for intrinsics setup to support this
     return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
    }
 
    //省略展示其它代碼...
}

這幾個(gè)方法都是Final修飾的，說明子類中無法重寫它們。我們可以通過修改State字段表示的同步狀態(tài)來實(shí)現(xiàn)多線程的獨(dú)占模式和共享模式（加鎖過程）。

2.2、等待隊(duì)列

等待隊(duì)列，是AQS框架的核心，整個(gè)框架的關(guān)鍵其實(shí)就是如何在并發(fā)狀態(tài)下管理被阻塞的線程。

等待隊(duì)列是嚴(yán)格的FIFO隊(duì)列，是Craig，Landin和Hagersten鎖（CLH鎖）的一種變種，采用雙向循環(huán)鏈表實(shí)現(xiàn)，因此也叫CLH隊(duì)列。

2.3、CLH隊(duì)列中的結(jié)點(diǎn)

AQS內(nèi)部還定義了一個(gè)靜態(tài)類Node，表示CLH隊(duì)列的每一個(gè)結(jié)點(diǎn)，該結(jié)點(diǎn)的作用是對(duì)每一個(gè)等待獲取資源做了封裝，包含了需要同步的線程本身、線程等待狀態(tài)....

LH隊(duì)列中的結(jié)點(diǎn)是對(duì)線程的包裝，結(jié)點(diǎn)一共有兩種類型：獨(dú)占（EXCLUSIVE）和共享（SHARED）。

每種類型的結(jié)點(diǎn)都有一些狀態(tài)，其中獨(dú)占結(jié)點(diǎn)使用其中的CANCELLED(1)、SIGNAL(-1)、CONDITION(-2)，共享結(jié)點(diǎn)使用其中的CANCELLED(1)、SIGNAL(-1)、PROPAGATE(-3)。

2.4、隊(duì)列定義

對(duì)于CLH隊(duì)列，當(dāng)線程請(qǐng)求資源時(shí)，如果請(qǐng)求不到，會(huì)將線程包裝成結(jié)點(diǎn)，將其掛載在隊(duì)列尾部。

下面結(jié)合代碼一起看下節(jié)點(diǎn)進(jìn)入隊(duì)列的過程。

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

2.5、AQS底層的CAS機(jī)制

在研究JDK中AQS時(shí)，會(huì)發(fā)現(xiàn)這個(gè)類很多地方都使用了CAS操作，在并發(fā)實(shí)現(xiàn)中CAS操作必須具備原子性，而且是硬件級(jí)別的原子性，Java被隔離在硬件之上，明顯力不從心，這時(shí)為了能直接操作操作系統(tǒng)層面，肯定要通過用C++編寫的native本地方法來擴(kuò)展實(shí)現(xiàn)。JDK提供了一個(gè)類來滿足CAS的要求，sun.misc.Unsafe，從名字上可以大概知道它用于執(zhí)行低級(jí)別、不安全的操作，AQS就是使用此類完成硬件級(jí)別的原子操作。UnSafe通過JNI調(diào)用本地C++代碼，C++代碼調(diào)用CPU硬件指令集。

Unsafe是一個(gè)很強(qiáng)大的類，它可以分配內(nèi)存、釋放內(nèi)存、可以定位對(duì)象某字段的位置、可以修改對(duì)象的字段值、可以使線程掛起、使線程恢復(fù)、可進(jìn)行硬件級(jí)別原子的CAS操作等等。

2.6、通過ReentrantLock理解AQS

ReentrantLock中公平鎖和非公平鎖在底層是相同的，這里以非公平鎖為例進(jìn)行分析。

在非公平鎖中，有一段這樣的代碼：

// java.util.concurrent.locks.ReentrantLock
 
static final class NonfairSync extends Sync {
    ...
    final void lock() {
        if (compareAndSetState(0, 1))
            setExclusiveOwnerThread(Thread.currentThread());
        else
            acquire(1);
    }
  ...
}

看一下這個(gè)Acquire是怎么寫的：

// java.util.concurrent.locks.AbstractQueuedSynchronizer
 
public final void acquire(int arg) {
    if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
}

再看一下tryAcquire方法：

// java.util.concurrent.locks.AbstractQueuedSynchronizer
 
protected boolean tryAcquire(int arg) {
    throw new UnsupportedOperationException();
}

可以看出，這里只是AQS的簡(jiǎn)單實(shí)現(xiàn)，具體獲取鎖的實(shí)現(xiàn)方法是由各自的公平鎖和非公平鎖單獨(dú)實(shí)現(xiàn)的（以ReentrantLock為例）。如果該方法返回了True，則說明當(dāng)前線程獲取鎖成功，就不用往后執(zhí)行了；如果獲取失敗，就需要加入到等待隊(duì)列中。

三、AQS方法

AQS代碼內(nèi)部提供了一系列操作鎖和線程隊(duì)列的方法，主要操作鎖的方法包含以下幾個(gè)：

compareAndSetState()：利用CAS的操作來設(shè)置state的值

tryAcquire(int)：獨(dú)占方式獲取鎖。成功則返回true，失敗則返回false。

tryRelease(int)：獨(dú)占方式釋放鎖。成功則返回true，失敗則返回false。

tryReleaseShared(int)：共享方式釋放鎖。如果釋放后允許喚醒后續(xù)等待結(jié)點(diǎn)返回true，否則返回false。

像ReentrantLock就是實(shí)現(xiàn)了自定義的tryAcquire-tryRelease，從而操作state的值來實(shí)現(xiàn)同步效果。

3.1、用戶需要自己重寫的方法

上面介紹到 AQS 已經(jīng)幫用戶解決了同步器定義過程中的大部分問題，只將下面兩個(gè)問題丟給用戶解決：

• 什么是資源
• 什么情況下資源是可以被訪問的

具體的，AQS 是通過暴露以下 API 來讓用戶解決上面的問題的。

如果你需要實(shí)現(xiàn)一個(gè)自己的同步器，一般情況下只要繼承 AQS ，并重寫 AQS 中的這個(gè)幾個(gè)方法就行了。至于具體線程等待隊(duì)列的維護(hù)（如獲取資源失敗入隊(duì)/喚醒出隊(duì)等），AQS已經(jīng)在頂層實(shí)現(xiàn)好了。要不怎么說Doug Lea貼心呢。

需要注意的是：如果你沒在子類中重寫這幾個(gè)方法就直接調(diào)用了，會(huì)直接拋出異常。所以，在你調(diào)用這些方法之前必須重寫他們。不使用的話可以不重寫。

3.2、AQS 提供的一系列模板方法

查看 AQS 的源碼我們就可以發(fā)現(xiàn)這個(gè)類提供了很多方法，看起來讓人“眼花繚亂”的。但是最主要的兩類方法就是獲取資源的方法和釋放資源的方法。因此我們抓住主要矛盾就行了：

• public final void acquire(int arg) // 獨(dú)占模式的獲取資源
• public final boolean release(int arg) // 獨(dú)占模式的釋放資源
• public final void acquireShared(int arg) // 共享模式的獲取資源
• public final boolean releaseShared(int arg) // 共享模式的釋放資源

3.3、acquire(int)方法

該方法以獨(dú)占方式獲取資源，如果獲取到資源，線程繼續(xù)往下執(zhí)行，否則進(jìn)入等待隊(duì)列，直到獲取到資源為止，且整個(gè)過程忽略中斷的影響。該方法是獨(dú)占模式下線程獲取共享資源的頂層入口。

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

下面分析下這個(gè)acquire方法的具體執(zhí)行流程：

step1：首先這個(gè)方法調(diào)用了用戶自己實(shí)現(xiàn)的方法tryAcquire方法嘗試獲取資源，如果這個(gè)方法返回true，也就是表示獲取資源成功，那么整個(gè)acquire方法就執(zhí)行結(jié)束了，線程繼續(xù)往下執(zhí)行；

step2：如果tryAcquir方法返回false，也就表示嘗試獲取資源失敗。這時(shí)acquire方法會(huì)先調(diào)用addWaiter方法將當(dāng)前線程封裝成Node類并加入一個(gè)FIFO的雙向隊(duì)列的尾部。

step3：再看acquireQueued這個(gè)關(guān)鍵方法。首先要注意的是這個(gè)方法中哪個(gè)無條件的for循環(huán)，這個(gè)for循環(huán)說明acquireQueued方法一直在自旋嘗試獲取資源。進(jìn)入for循環(huán)后，首先判斷了當(dāng)前節(jié)點(diǎn)的前繼節(jié)點(diǎn)是不是頭節(jié)點(diǎn)，如果是的話就再次嘗試獲取資源，獲取資源成功的話就直接返回false（表示未被中斷過）

假如還是沒有獲取資源成功，判斷是否需要讓當(dāng)前節(jié)點(diǎn)進(jìn)入waiting狀態(tài)，經(jīng)過 shouldParkAfterFailedAcquire這個(gè)方法判斷，如果需要讓線程進(jìn)入waiting狀態(tài)的話，就調(diào)用LockSupport的park方法讓線程進(jìn)入waiting狀態(tài)。進(jìn)入waiting狀態(tài)后，這線程等待被interupt或者unpark（在release操作中會(huì)進(jìn)行這樣的操作，可以參見后面的代碼）。這個(gè)線程被喚醒后繼續(xù)執(zhí)行for循環(huán)來嘗試獲取資源。

final boolean acquireQueued(final Node node, int arg) {
        boolean failed = true;
        try {
            boolean interrupted = false;
            for (;;) {
                final Node p = node.predecessor();
                //首先判斷了當(dāng)前節(jié)點(diǎn)的前繼節(jié)點(diǎn)是不是頭節(jié)點(diǎn)，如果是的話就再次嘗試獲取資源，
                //獲取資源成功的話就直接返回false（表示未被中斷過）
                if (p == head && tryAcquire(arg)) {
                    setHead(node);
                    p.next = null; // help GC
                    failed = false;
                    return interrupted;
                }
                //判斷是否需要讓當(dāng)前節(jié)點(diǎn)進(jìn)入waiting狀態(tài)
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    // 如果在整個(gè)等待過程中被中斷過，則返回true，否則返回false。
                    // 如果線程在等待過程中被中斷過，它是不響應(yīng)的。只是獲取資源后才再進(jìn)行自我中斷selfInterrupt()，將中斷補(bǔ)上。
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

以上就是acquire方法的簡(jiǎn)單分析。

單獨(dú)看這個(gè)方法的話可能會(huì)不太清晰，結(jié)合ReentrantLock、ReentrantReadWriteLock、CountDownLatch、Semaphore和LimitLatch等同步工具看這個(gè)代碼的話就會(huì)好理解很多。

3.4、release(int)方法

release(int)方法是獨(dú)占模式下線程釋放共享資源的頂層入口。它會(huì)釋放指定量的資源，如果徹底釋放了（即state=0）,它會(huì)喚醒等待隊(duì)列里的其他線程來獲取資源。

public final boolean release(int arg) {
    if (tryRelease(arg)) {
        Node h = head;
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        return true;
    }
    return false;
}
 
//上面已經(jīng)講過了，需要用戶自定義實(shí)現(xiàn)
protected boolean tryRelease(int arg) {
    throw new UnsupportedOperationException();
}
 
private void unparkSuccessor(Node node) {
    /*
         * If status is negative (i.e., possibly needing signal) try
         * to clear in anticipation of signalling.  It is OK if this
         * fails or if status is changed by waiting thread.
         */
    int ws = node.waitStatus;
    if (ws < 0)
        compareAndSetWaitStatus(node, ws, 0);
 
    /*
         * Thread to unpark is held in successor, which is normally
         * just the next node.  But if cancelled or apparently null,
         * traverse backwards from tail to find the actual
         * non-cancelled successor.
         */
    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);
}

與acquire()方法中的tryAcquire()類似，tryRelease()方法也是需要獨(dú)占模式的自定義同步器去實(shí)現(xiàn)的。正常來說，tryRelease()都會(huì)成功的，因?yàn)檫@是獨(dú)占模式，該線程來釋放資源，那么它肯定已經(jīng)拿到獨(dú)占資源了，直接減掉相應(yīng)量的資源即可(state-=arg)，也不需要考慮線程安全的問題。

但要注意它的返回值，上面已經(jīng)提到了，release()是根據(jù)tryRelease()的返回值來判斷該線程是否已經(jīng)完成釋放掉資源了！所以自義定同步器在實(shí)現(xiàn)時(shí)，如果已經(jīng)徹底釋放資源(state=0)，要返回true，否則返回false。

unparkSuccessor(Node)方法用于喚醒等待隊(duì)列中下一個(gè)線程。這里要注意的是，下一個(gè)線程并不一定是當(dāng)前節(jié)點(diǎn)的next節(jié)點(diǎn)，而是下一個(gè)可以用來喚醒的線程，如果這個(gè)節(jié)點(diǎn)存在，調(diào)用unpark()方法喚醒。

總之，release()是獨(dú)占模式下線程釋放共享資源的頂層入口。它會(huì)釋放指定量的資源，如果徹底釋放了（即state=0）,它會(huì)喚醒等待隊(duì)列里的其他線程來獲取資源。（需要注意的是隊(duì)列中被喚醒的線程不一定能立馬獲取資源，因?yàn)橘Y源在釋放后可能立馬被其他線程（不是在隊(duì)列中等待的線程）搶掉了）

3.5、acquireShared(int)方法

acquireShared(int)方法是共享模式下線程獲取共享資源的頂層入口。它會(huì)獲取指定量的資源，獲取成功則直接返回，獲取失敗則進(jìn)入等待隊(duì)列，直到獲取到資源為止，整個(gè)過程忽略中斷。

public final void acquireShared(int arg) {
    //tryAcquireShared需要用戶自定義實(shí)現(xiàn)
    if (tryAcquireShared(arg) < 0)
        doAcquireShared(arg);
}

可以發(fā)現(xiàn)，這個(gè)方法的關(guān)鍵實(shí)現(xiàn)其實(shí)是獲取資源失敗后，怎么管理線程。也就是doAcquireShared的邏輯。

//不響應(yīng)中斷
private void doAcquireShared(int arg) {
    final Node node = addWaiter(Node.SHARED);
    boolean failed = true;
    try {
        boolean interrupted = false;
        for (;;) {
            final Node p = node.predecessor();
            if (p == head) {
                int r = tryAcquireShared(arg);
                if (r >= 0) {
                    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);
    }
}

可以看出，doAcquireShared的邏輯和acquireQueued的邏輯差不多。將當(dāng)前線程加入等待隊(duì)列尾部休息，直到其他線程釋放資源喚醒自己，自己成功拿到相應(yīng)量的資源后才返回。

簡(jiǎn)單總結(jié)下acquireShared的流程：

step1：tryAcquireShared()嘗試獲取資源，成功則直接返回；

step2：失敗則通過doAcquireShared()進(jìn)入等待隊(duì)列park()，直到被unpark()/interrupt()并成功獲取到資源才返回。整個(gè)等待過程也是忽略中斷的。

3.6、releaseShared(int)方法

releaseShared(int)方法是共享模式下線程釋放共享資源的頂層入口。它會(huì)釋放指定量的資源，如果成功釋放且允許喚醒等待線程，它會(huì)喚醒等待隊(duì)列里的其他線程來獲取資源。

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

釋放掉資源后，喚醒后繼。跟獨(dú)占模式下的release()相似，但有一點(diǎn)稍微需要注意：獨(dú)占模式下的tryRelease()在完全釋放掉資源（state=0）后，才會(huì)返回true去喚醒其他線程，這主要是基于獨(dú)占下可重入的考量；而共享模式下的releaseShared()則沒有這種要求，共享模式實(shí)質(zhì)就是控制一定量的線程并發(fā)執(zhí)行，那么擁有資源的線程在釋放掉部分資源時(shí)就可以喚醒后繼等待結(jié)點(diǎn)。

參考鏈接：

從ReentrantLock的實(shí)現(xiàn)看AQS的原理及應(yīng)用

并發(fā)編程的基石——AQS類

總結(jié)

本篇文章就到這里了，希望能夠給你帶來幫助，也希望您能夠多多關(guān)注腳本之家的更多內(nèi)容！

最新評(píng)論