一、原理概述

AQS全稱是 AbstractQueuedSynchronizer，是阻塞式鎖和相關(guān)的同步器工具的框架

AQS核心思想是，如果被請(qǐng)求的共享資源（state）空閑，則將當(dāng)前請(qǐng)求資源的線程設(shè)置為有效的工作線程，并且將共享資源設(shè)置為鎖定狀態(tài)。如果被請(qǐng)求的共享資源被占用，那么就需要一套線程阻塞等待以及被喚醒時(shí)鎖分配的機(jī)制，這個(gè)機(jī)制AQS是用CLH隊(duì)列鎖實(shí)現(xiàn)的，即將暫時(shí)獲取不到鎖的線程加入到隊(duì)列中。

CLH隊(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(AbstractQueuedSynchronizer)原理圖：

AQS使用一個(gè)int成員變量來表示同步狀態(tài)，通過內(nèi)置的FIFO隊(duì)列來完成獲取資源線程的排隊(duì)工作。AQS使用CAS對(duì)該同步狀態(tài)進(jìn)行原子操作實(shí)現(xiàn)對(duì)其值的修改。

//共享變量，使用volatile修飾保證線程可見性
private volatile int state;
//狀態(tài)信息通過protected類型的getState，setState，compareAndSetState進(jìn)行操作. 且為final類型，不允許被子類重寫
//返回同步狀態(tài)的當(dāng)前值
protected final int getState() {
        return state;
}
 // 設(shè)置同步狀態(tài)的值
protected final void setState(int newState) {
        state = newState;
}
//原子地（CAS操作）將同步狀態(tài)值設(shè)置為給定值update如果當(dāng)前同步狀態(tài)的值等于expect（期望值）
protected final boolean compareAndSetState(int expect, int update) {
        return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}

二、AQS 對(duì)資源的共享方式

AQS定義兩種資源共享方式

• Exclusive（獨(dú)占）：只有一個(gè)線程能執(zhí)行，如ReentrantLock。又可分為公平鎖和非公平鎖： - 公平鎖：按照線程在隊(duì)列中的排隊(duì)順序，先到者先拿到鎖 - 非公平鎖：當(dāng)線程要獲取鎖時(shí)，無視隊(duì)列順序直接去搶鎖，誰搶到就是誰的
• share（共享）：多個(gè)線程都可以同時(shí)獲取到鎖，如Semaphore/CountDownLatch。Semaphore、CountDownLatch、 CyclicBarrier、ReadWriteLock 我們都會(huì)在后面講到。

不同的自定義同步器爭(zhēng)用共享資源的方式也不同。自定義同步器在實(shí)現(xiàn)時(shí)只需要實(shí)現(xiàn)共享資源 state 的獲取與釋放方式即可，至于具體線程等待隊(duì)列的維護(hù)（如獲取資源失敗入隊(duì)/喚醒出隊(duì)等），AQS已經(jīng)在頂層實(shí)現(xiàn)好了

三、AQS底層使用了模板方法模式

同步器的設(shè)計(jì)是基于模板方法模式的，如果需要自定義同步器一般的方式是這樣

• 自定義同步器繼承AbstractQueuedSynchronizer并重寫指定的方法。（這些重寫方法很簡單，無非是對(duì)于共享資源state的獲取和釋放）
• 將自定義同步器組合在自定義同步組件的實(shí)現(xiàn)中，并調(diào)用其模板方法，而這些模板方法會(huì)調(diào)用使用者重寫的方法

自定義同步器時(shí)需要重寫下面幾個(gè)AQS提供的模板方法：

isHeldExclusively()//該線程是否正在獨(dú)占資源。只有用到condition才需要去實(shí)現(xiàn)它。
tryAcquire(int)//獨(dú)占方式。嘗試獲取資源，成功則返回true，失敗則返回false。
tryRelease(int)//獨(dú)占方式。嘗試釋放資源，成功則返回true，失敗則返回false。
tryAcquireShared(int)//共享方式。嘗試獲取資源。負(fù)數(shù)表示失??；0表示成功，但沒有剩余可用資源；正數(shù)表示成功，且有剩余資源。
tryReleaseShared(int)//共享方式。嘗試釋放資源，成功則返回true，失敗則返回false。

默認(rèn)情況下，每個(gè)方法都拋出 UnsupportedOperationException。這些方法的實(shí)現(xiàn)必須是內(nèi)部線程安全的，并且通常應(yīng)該簡短而不是阻塞。

AQS類中的其他方法都是final ，所以無法被其他類使用，只有這幾個(gè)方法可以被其他類使用。

以ReentrantLock為例，state初始化為0，表示未鎖定狀態(tài)。A線程lock()時(shí)，會(huì)調(diào)用tryAcquire()獨(dú)占該鎖并將state+1。此后，其他線程再tryAcquire()時(shí)就會(huì)失敗，直到A線程unlock()到state=0（即釋放鎖）為止，其它線程才有機(jī)會(huì)獲取該鎖。當(dāng)然，釋放鎖之前，A線程自己是可以重復(fù)獲取此鎖的（state會(huì)累加），這就是可重入的概念。但要注意，獲取多少次就要釋放多么次，這樣才能保證state是能回到零態(tài)的。

再以CountDownLatch以例，任務(wù)分為N個(gè)子線程去執(zhí)行，state也初始化為N（注意N要與線程個(gè)數(shù)一致）。這N個(gè)子線程是并行執(zhí)行的，每個(gè)子線程執(zhí)行完后countDown()一次，state會(huì)CAS(Compare and Swap)減1。等到所有子線程都執(zhí)行完后(即state=0)，會(huì)unpark()主調(diào)用線程，然后主調(diào)用線程就會(huì)從await()函數(shù)返回，繼續(xù)后余動(dòng)作。

一般來說，自定義同步器要么是獨(dú)占方法，要么是共享方式，他們也只需實(shí)現(xiàn)tryAcquire-tryRelease、tryAcquireShared-tryReleaseShared中的一種即可。但AQS也支持自定義同步器同時(shí)實(shí)現(xiàn)獨(dú)占和共享兩種方式，如ReentrantReadWriteLock。

四、使用demo，使用AQS實(shí)現(xiàn)不可重入鎖

實(shí)現(xiàn)不可重入鎖需要分兩步來走，一是實(shí)現(xiàn)自定義同步器，二是實(shí)現(xiàn)自定義鎖

自定義同步器

class MySync extends AbstractQueuedSynchronizer {
        // 嘗試去獲取鎖
        @Override
        protected boolean tryAcquire(int i) {
            // compareAndSetState(0, 1): 嘗試著將state的值從 0改為1
            if (compareAndSetState(0, 1)) {
                // 將持有鎖的線程改為當(dāng)前線程
                setExclusiveOwnerThread(Thread.currentThread());
                return true;
            }
            return false;
        }
        // 嘗試去釋放鎖
        @Override
        protected boolean tryRelease(int i) {
            setExclusiveOwnerThread(null);
            setState(0);
            return true;
        }
        // 判斷是否持有獨(dú)占鎖
        @Override
        protected boolean isHeldExclusively() {
            return getState() == 1;
        }
        protected Condition newCondition() {
            return new ConditionObject();
        }
    }

自定義鎖

有了自定義同步器，很容易復(fù)用 AQS ，實(shí)現(xiàn)一個(gè)功能完備的自定義鎖

public class MyLock implements Lock {
    // 自定義同步器 , 實(shí)現(xiàn)的是不可重入鎖
    private MySync sync = new MySync();
    @Override
    // 嘗試，不成功，進(jìn)入等待隊(duì)列
    public void lock() {
        sync.acquire(1);
    }
    @Override
    // 嘗試，不成功，進(jìn)入等待隊(duì)列，可打斷
    public void lockInterruptibly() throws InterruptedException {
        sync.acquireInterruptibly(1);
    }
    @Override
    // 嘗試一次，不成功返回，不進(jìn)入隊(duì)列
    public boolean tryLock() {
        return sync.tryAcquire(1);
    }
    @Override
    // 嘗試，不成功，進(jìn)入等待隊(duì)列，有時(shí)限
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        return sync.tryAcquireNanos(1, unit.toNanos(time));
    }
    @Override
    // 釋放鎖
    public void unlock() {
        sync.release(1);
    }
    @Override
    // 生成條件變量
    public Condition newCondition() {
        return sync.newCondition();
    }
}

測(cè)試

@Slf4j(topic = "c.MyLockTest")
public class MyLockTest {
    public static void main(String[] args) {
        MyLock lock = new MyLock();
        new Thread(() -> {
            lock.lock();
            try {
                log.debug("locking...");
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            } finally {
                log.debug("unlocking...");
                lock.unlock();
            }
        },"t1").start();
        new Thread(() -> {
            lock.lock();
            try {
                log.debug("locking...");
            } finally {
                log.debug("unlocking...");
                lock.unlock();
            }
        },"t2").start();
    }
}

可以看到只有當(dāng)t1線程釋放鎖之后，t2線程才能獲取到鎖

五、AQS使用到的幾個(gè)框架

1、信號(hào)量 Semaphore 允許多個(gè)線程同時(shí)訪問： synchronized 和 ReentrantLock 都是一次只允許一個(gè)線程訪問某個(gè)資源，Semaphore(信號(hào)量)可以指定多個(gè)線程同時(shí)訪問某個(gè)資源。

2、CountDownLatch （倒計(jì)時(shí)器） CountDownLatch是一個(gè)同步工具類，用來協(xié)調(diào)多個(gè)線程之間的同步。這個(gè)工具通常用來控制線程等待，它可以讓某一個(gè)線程等待直到倒計(jì)時(shí)結(jié)束，再開始執(zhí)行。

3、CyclicBarrier(循環(huán)柵欄) CountDownLatch 更加復(fù)雜和強(qiáng)大。主要應(yīng)用場(chǎng)景和 CountDownLatch 類似。CyclicBarrier 的字面意思是可循環(huán)使用（Cyclic）的屏障（Barrier）。它要做的事情是，讓一組線程到達(dá)一個(gè)屏障（也可以叫同步點(diǎn)）時(shí)被阻塞，直到最后一個(gè)線程到達(dá)屏障時(shí)，屏障才會(huì)開門，所有被屏障攔截的線程才會(huì)繼續(xù)干活。CyclicBarrier默認(rèn)的構(gòu)造方法是 CyclicBarrier(int parties)，其參數(shù)表示屏障攔截的線程數(shù)量，每個(gè)線程調(diào)用await()方法告訴 CyclicBarrier 我已經(jīng)到達(dá)了屏障，然后當(dāng)前線程被阻塞。

到此這篇關(guān)于Java中的AQS框架原理詳解的文章就介紹到這了,更多相關(guān)AQS框架原理內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論