AQS 同步隊列是很多的 Java 線程安全對象的實現(xiàn)，例如 ReentrantLock, Semaphore, CountDownLatch, ReentrantReadWriteLock 等等。

AQS 是 AbstractQueuedSynchronizer 的簡稱，它是一個抽象類，我們需要實現(xiàn)其中的一些關(guān)鍵方法來完成他的基本功能。

這里簡單介紹一下它的實現(xiàn)方式，當一個線程想要獲取該對象的鎖的時候，會通過方法檢查該線程是否能夠獲取鎖，如果能夠獲取鎖就結(jié)束了，完事兒；如果不能夠獲取鎖，就加入同步隊列等待，同時掛起該線程，如果這個時候還有別的線程在競爭該對象的鎖接著加入同步隊列，掛起，當占有這個鎖的線程完事兒后會釋放鎖，釋放時會去檢查同步隊列，取出最先進入隊列的線程，然后把它喚醒，它就獲得了鎖，當它也完事兒釋放后，又喚醒下一個，直到隊列中的等待線程全部喚醒。

網(wǎng)上已經(jīng)有很多的源碼分析的文章了，所以我想盡可能的簡化分析，很多的細節(jié)我就不說了。

1 自定義 AQS 的重要方法

val qs = object : AbstractQueuedSynchronizer() {
    /**
     * 嘗試獲取互斥鎖，如果返回，如果獲取失敗，后續(xù)就會進入同步隊列，同時掛起線程
     */
    override fun tryAcquire(arg: Int): Boolean {
        return super.tryAcquire(arg)
    }
    /**
     * 嘗試釋放互斥鎖
     */
    override fun tryRelease(arg: Int): Boolean {
        return super.tryRelease(arg)
    }
    /**
     * 嘗試獲取共享鎖，和同步鎖一樣，失敗就進入隊列
     */
    override fun tryAcquireShared(arg: Int): Int {
        return super.tryAcquireShared(arg)
    }
    /**
     * 嘗試釋放同步鎖
     */
    override fun tryReleaseShared(arg: Int): Boolean {
        return super.tryReleaseShared(arg)
    }
    /**
     * 當前線程是否獲得鎖
     */
    override fun isHeldExclusively(): Boolean {
        return super.isHeldExclusively()
    }
}

下面是 JDK 中 ReentrantLock 中不公平鎖的實現(xiàn)：

? ? class Sync extends AbstractQueuedSynchronizer {
? ? ? ? private static final long serialVersionUID = -5179523762034025860L;
? ? ? ? /**
? ? ? ? ?* Performs non-fair tryLock. ?tryAcquire is implemented in
? ? ? ? ?* subclasses, but both need nonfair try for trylock method.
? ? ? ? ?*/
? ? ? ? @ReservedStackAccess
? ? ? ? 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 tryAcquire(int acquires) {
? ? ? ? ? ? return nonfairTryAcquire(acquires);
? ? ? ? }
? ? ? ? @ReservedStackAccess
? ? ? ? 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() {
? ? ? ? ? ? // While we must in general read state before owner,
? ? ? ? ? ? // we don't need to do so to check if current thread is owner
? ? ? ? ? ? return getExclusiveOwnerThread() == Thread.currentThread();
? ? ? ? }
? ? ? ? final ConditionObject newCondition() {
? ? ? ? ? ? return new ConditionObject();
? ? ? ? }
? ? ? ? // Methods relayed from outer class
? ? ? ? final Thread getOwner() {
? ? ? ? ? ? return getState() == 0 ? null : getExclusiveOwnerThread();
? ? ? ? }
? ? ? ? final int getHoldCount() {
? ? ? ? ? ? return isHeldExclusively() ? getState() : 0;
? ? ? ? }
? ? ? ? final boolean isLocked() {
? ? ? ? ? ? return getState() != 0;
? ? ? ? }
? ? }

ReentrantLock 只實現(xiàn)了互斥鎖。

• 嘗試獲取互斥鎖

acquires 這個值外部每次都是傳的 1，首先通過 getState() 方法獲取 state。
如果 state 為 0 就表示鎖可以使用，這時通過 CAS 的方式設(shè)置新的狀態(tài)，如果 CAS 競爭失敗，說明有其他線程同時也在競爭這個鎖，這時就直接返回失敗，如果競爭成功就會通過setExclusiveOwnerThread 方法設(shè)置當前線程為擁有者，返回成功。
如果 state 不為 0 但是 owner 就是為當前線程，就表示當前線程調(diào)用了多次 lock() 方法，這次就是簡單的在原有的 state 上再加 1，同時返回獲取鎖成功。
其他的情況也就是返回失敗了，失敗了就會和前面說到的就會進入同步隊列，同時掛起當前線程。

• 嘗試釋放互斥鎖

同樣的 releases 每次傳的值也是 1。
首先判斷當前線程是否擁有鎖，如果不擁有鎖直接拋出異常，通過 getState() 獲取上次的狀態(tài)，上次的狀態(tài)再減去 releases，就是新的狀態(tài)，如果新的狀態(tài)為 0 就表示應(yīng)該釋放鎖，同時通過 setExclusiveOwnerThread 方法把擁有線程設(shè)置為空，同時返回 true，其他情況返回 false。
根據(jù)前面介紹的，如果成功釋放，這時 AQS 還得去檢查同步隊列，拿到最近一個等待鎖的線程，并喚醒。

上面鎖的代碼很簡單，相信你已經(jīng)看懂了，這也解釋了為什么叫可重入鎖，也就是同一個線程可以多次調(diào)用 lock() 方法，同樣的也要對等的調(diào)用 unlock() 方法完成解鎖，lock() 和 unlock() 必須成對出現(xiàn)。共享鎖的實現(xiàn)我就不貼了，大同小異。

2 互斥鎖

2.1 沒有出現(xiàn)鎖競爭

前面的 Reentrant 源碼也提到了，在沒有鎖競爭的時候和被其他線程占有鎖的情況下，只是簡單的設(shè)置 state 為 1 和 設(shè)置 owner 的線程。
這個過程的性能消耗是非常小的，幾乎可以忽略不計。

2.2 鎖競爭失敗或者鎖已經(jīng)被占用

如果嘗試獲取鎖失敗，就會新建一個 Node 對象（AQS 的隊列實現(xiàn)是雙向鏈表，Node 就是他的節(jié)點實現(xiàn)），其中包含了節(jié)點狀態(tài)和關(guān)聯(lián)線程等關(guān)鍵信息，創(chuàng)建后將其加入到等待隊列的隊尾，同時將其線程掛起（掛起是使用的 LockSupport.park() 方法，其內(nèi)部的實現(xiàn)是 C++ 代碼，使用的是 Mutex 和 synchronized 用的是一樣的方法），需要等待占用鎖的線程釋放鎖后，根據(jù)同步隊列的順序把下一個同步隊列的線程喚醒（喚醒使用的是 LockSupport.unpark() 方法）。

這種情況和沒有鎖競爭的情況性能消耗就要大一些了。在進入隊列和釋放鎖的過程中可能會有多次的 CAS 自旋（也就是 CAS 失敗后通過 while 循環(huán)重試，直到成功，這時 CPU 是在空轉(zhuǎn)）；還有關(guān)鍵一點是線程的掛起和喚醒是需要操作系統(tǒng)來操作的，也就是會涉及到用戶態(tài)向內(nèi)核態(tài)的轉(zhuǎn)換，這個過程是比較消耗性能的。

3 共享鎖

共享鎖如果在理解了互斥鎖的前提下是比較簡單的。

在沒有被互斥鎖占用的情況下（tryAcquireShared() 方法返回 true），共享鎖是每一個線程都不會被掛起。

在互斥鎖被占用的情況下(tryAcquireShared() 方法返回 false)，也會創(chuàng)建一個 Node 對象加入到隊列中，不過添加了一個 waiter 對象來標記這是一個共享的節(jié)點，同樣的這個線程也會被掛起，等待互斥鎖被釋放后，按照先后會喚醒該線程，當該線程被喚醒后如果他的下一個節(jié)點也是共享的節(jié)點也會被喚醒，就像多米諾骨牌一樣挨個喚醒所有的共享節(jié)點，直到又被下一個互斥結(jié)點把互斥鎖給占用。

4 Condition

AQS 的 Condition 的 await/signal 和 synchronized 的 wait/notify 有異曲同工之妙。在獲取到互斥鎖后可以通過 Condition#await 方法把鎖釋放出去，同時自己被掛起，當獲取到鎖的線程調(diào)用 Condition#signal 方法又可以喚醒之前 await 掛起的線程。

在每個 Condition 對象中也會維護一個隊列（和 AQS 中的隊列是分開的，但是都是 Node 對象），每次有獲取鎖的線程調(diào)用 await 方法后都會在其中添加一個 Node，會用 CONDITION 標注狀態(tài)，同時釋放當前占用的鎖喚醒同步隊列中的下一個線程，并把自己掛起。當有線程調(diào)用 signal 方法后，會把 Condition 對象中的頭節(jié)點(如果是 signalAll 就是把全部的節(jié)點都加入到 AQS 隊列中)加入到 AQS 的同步隊列中，同時觸發(fā) await 方法重新去獲取鎖，這里也和前面說的獲取同步鎖一樣，就相當于 signal 后，await 需要重新排隊去獲取互斥鎖。

5 最后

最后再簡單聊一下 synchronized 和 AQS, 在 Java 早期的版本，每次 synchronized 都會去請求 mutex，導(dǎo)致沒有鎖競爭的時候性能不好，在 1.6 版本后加入了多級鎖，性能得到了不錯的提升。
在 Java 對象中定義了一個 Mark Word 空間來記錄對象的一些信息，其中就包括了重要的鎖信息，在對象沒有鎖的時候，一個線程需獲取鎖默認就是偏向鎖, 只需要在對象的 Mark Word 中通過 CAS 設(shè)置鎖的類型和鎖屬于的線程 ID，當沒有別的線程競爭那就皆大歡喜，完事兒了；如果在 偏向鎖競爭失敗或者占有偏向鎖的線程還沒有完事兒，那么鎖就會升級成輕量鎖，當然升級后還是之前持有偏向鎖的線程繼續(xù)持有，其中輕量鎖需要在持有的線程中添加一個 Lock Record 來指向?qū)?yīng)的對象，對象的 Mark Work 也會添加指向 Thread 對應(yīng)的 Lock Record，在等待獲取鎖的線程也會通過 CAS 自旋的方式去修改這些值，來嘗試獲取輕量鎖，當自旋超過一定次數(shù)了或者有別的線程來競爭，這時就會升級成 重量鎖，重量鎖也是用了 monitor 鎖，內(nèi)部也是用 mutex 實現(xiàn)。
synchronized 和 AQS 目前在性能上差距不大，當有多個線程競爭是都會升級成 mutex，不同的是 synchronized 使用起來非常簡單，但是功能沒有那么多，AQS 使用起來比較復(fù)雜，但是包含互斥鎖和共享鎖，他們之間的組合能夠完成很多復(fù)雜的功能，JDK 中很多的線程安全對象也用到了 AQS。

到此這篇關(guān)于Java AQS 線程安全同步隊列的實現(xiàn)的文章就介紹到這了,更多相關(guān)Java AQS 線程安全同步隊列內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評論