在多線程編程中我們會遇到很多需要使用線程同步機制去解決的并發問題，而這些同步機制就是多線程編程中影響正確性和運行效率的重中之重。這不禁讓我感到好奇，這些同步機制是如何實現的呢？好奇心是進步的源泉，就讓我們一起來揭開同步機制源碼的神秘面紗吧。

在本文中，我們會從JDK中大多數同步機制的共同基礎AbstractQueuedSynchronizer類開始說起，然后通過源碼了解我們最常用的兩個同步類可重入鎖ReentrantLock和閉鎖CountDownLatch的具體實現。通過這篇文章我們將可以了解到ReentrantLock和CountDownLatch兩個常用同步類的源代碼實現，并且掌握閱讀其他基于AQS實現的同步工具類源碼的能力，甚至可以利用AQS寫出自己的同步工具類。

閱讀這篇文章需要了解基本的線程同步機制，有興趣的讀者可以參考一下這篇文章《多線程中那些看不到的陷阱》。

同步機制的核心——AQS

同步機制源碼初探

ReentrantLock是我們常用的一種可重入互斥鎖，是synchronized關鍵字的一個很好的替代品。互斥指的就是同一時間只能有一個線程獲取到這個鎖，而可重入是指如果一個線程再次獲取一個它已經持有的互斥鎖，那么仍然會成功。

這個類的源碼在JDK的java.util.concurrent包下，我們可以在IDE中點擊類名跳轉到具體的類定義，比如下面就是在我的電腦上跳轉之后看到的ReentrantLock類的源代碼。在這里我們可以看到在ReentrantLock類中還包含了一個繼承自AbstractQueuedSynchronizer類的內部類，而且有一個該內部類Sync類型的字段sync。實際上ReentrantLock類就是通過這個內部類對象來實現線程同步的。

如果打開CountDownLatch的源代碼，我們會發現這個類里也同樣有一個繼承自AbstractQueuedSynchronizer類的子類Sync，并且也有一個Sync類型的字段sync。在java.util.concurrent包下的大多數同步工具類的底層都是通過在內部定義一個AbstractQueuedSynchronizer類的子類來實現的，包括我們在本文中沒提到的許多其他常用類也是如此，比如：讀寫鎖ReentrantReadWriteLock、信號量Semaphore等。

AQS是什么？

那么這個AbstractQueuedSynchronizer類也就是我們所說的AQS，到底是何方神圣呢？這個類首先像我們上面提到的，是大多數多線程同步工具類的基礎。它內部包含了一個對同步器的等待隊列，其中包含了所有在等待獲取同步器的線程，在這個等待隊列中的線程將會在同步器釋放時被喚醒。比如一個線程在獲取互斥鎖失敗時就會被放入到等待隊列中等待被喚醒，這也就是AQS中的Q——“Queued”的由來。

而類名中的***個單詞Abstract是因為AQS是一個抽象類，它的使用方法就是實現繼承它的子類，然后使用這個子類類型的對象。在這個子類中我們會通過重寫下列的五個方法中的一部分或者全部來指定這個同步器的行為策略：

1.  boolean tryAcquire(int arg)，獨占式獲取同步器，獨占式指同一時間只能有一個線程獲取到同步器；
2.  boolean tryRelease(int arg)，獨占式釋放同步器；
3.  boolean isHeldExclusively()，同步器是否被當前線程獨占式地持有；
4.  int tryAcquireShared(int arg)，共享式獲取同步器，共享式指的是同一時間可能有多個線程同時獲取到同步器，但是可能會有數量的限制；
5.  boolean tryReleaseShared(int arg)，共享式釋放同步器。

這五個方法之所以能指定同步器的行為，則是因為AQS中的其他方法就是通過對這五個方法的調用來實現的。比如在下面的acquire方法中就調用了tryAcquire來獲取同步器，并且在被調用的acquireQueued方法內部也是通過tryAcquire方法來循環嘗試獲取同步器的。 

public final void acquire(int arg) {  
    // 1. 調用tryAcquire方法嘗試獲取鎖  
    // 2. 如果獲取失敗(tryAcquire返回false)，則調用addWaiter方法將當前線程保存到等待隊列中  
    // 3. 之后調用acquireQueued方法來循環執行“獲取同步器 -> 獲取失敗休眠 -> 被喚醒重新獲取”過程  
    //      直到成功獲取到同步器返回false；或者被中斷返回true  
    if (!tryAcquire(arg) &&  
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))  
        // 如果acquireQueued方法返回true說明線程被中斷了  
        //   所以調用selfInterrupt方法中斷當前線程  
        selfInterrupt();  
} 

下面，我們就來看看在ReentrantLock和CountDownLatch兩個類中定義的AQS子類到底是如何重寫這五個方法的。

CountDownLatch的實現

CountDownLatch是一種典型的閉鎖，比如我需要使用四個線程完成四種不同的計算，然后把四個線程的計算結果相加后返回，這種情況下主線程就需要等待四個完成不同任務的工作線程完成之后才能繼續執行。那么我們就可以創建一個初始的count值為4的CountDownLatch，然后在每個工作線程完成任務時都對這個CountDownLatch執行一個countDown操作，這樣CountDownLatch中的count值就會減1。當count值減到0時，主線程就會從阻塞中恢復，然后將四個任務的結果相加后返回。

下面是CountDownLath的幾個常用方法：

1.  void await()，等待操作，如果count值目前已經是0了，那么就直接返回；否則就進入阻塞狀態，等待count值變為0；
2.  void countDown()，減少計數操作，會讓count減1。

調用多次countDown()方法讓count值變為0之后，被await()方法阻塞的線程就可以繼續執行了。了解了CountDownLatch的基本用法之后我們就來看看這個閉鎖到底是怎么實現的，首先，我們來看一下CountDownLatch中AQS的子類，內部類Sync的定義。

CountDownLatch的內部Sync類

下面的代碼是CountDownLatch中AQS的子類Sync的定義，Sync是CountDownLatch類中的一個內部類。在這個類中重寫了AQS的tryAcquireShared和tryReleaseShared兩個方法，這兩個都是共享模式需要重寫的方法，因為CountDownLatch在count值為0時可以被任意多個線程同時獲取成功，所以應該實現共享模式的方法。

在CountDownLatch的Sync中使用了AQS的state值用來存放count值，在初始化時會把state值初始化為n。然后在調用tryReleaseShared時會將count值減1，但是因為這個方法可能會被多個線程同時調用，所以要用CAS操作保證更新操作的原子性，就像我們用AtomicInteger一樣。在CAS失敗時我們需要通過重試來保證把state減1，如果CAS成功時，即使有許多線程同時執行這個操作***的結果也一定是正確的。在這里，tryReleaseShared方法的返回值表示這個釋放操作是否可以讓等待中的線程成功獲取同步器，所以只有在count為0時才能返回true。

tryAcquireShared方法就比較簡單了，直接返回state是否等于0即可，因為只有在CountDownLatch中的count值為0時所有希望獲取同步器的線程才能獲取成功并繼續執行。如果count不為0，那么線程就需要進入阻塞狀態，等到count值變為0才能繼續執行。 

private static final class Sync extends AbstractQueuedSynchronizer {  
    private static final long serialVersionUID = 4982264981922014374L;  
    // 構造器，初始化count值  
    // 在這個子類中把count值保存到了AQS的state中  
    Sync(int count) {  
        setState(count);  
    }  
    // 獲取當前的count值  
    int getCount() {  
        return getState();  
    }  
    // 獲取操作在state為0時會成功，否則失敗  
    // tryAcquireShared失敗時，線程會進入阻塞狀態等待獲取成功  
    protected int tryAcquireShared(int acquires) {  
        return (getState() == 0) ? 1 : -1;  
    }  
    // 對閉鎖執行釋放操作減小計數值  
    protected boolean tryReleaseShared(int releases) {  
        // 減小coun值，在count值歸零時喚醒等待的線程  
        for (;;) {  
            int c = getState();  
            // 如果計數已經歸零，則直接釋放失敗  
            if (c == 0)  
                return false;  
            // 將計數值減1  
            int nextc = c-1;  
            // 為了線程安全，以CAS循環嘗試更新  
            if (compareAndSetState(c, nextc))  
                return nextc == 0;  
        }  
    }  
} 

CounDownLatch對Sync類對象的使用

看了CountDownLatch中的Sync內部類定義之后，我們再來看看CountDownLatch是如何使用這個內部類的。

在CountDownLatch的構造器中，初始化CountDownLatch對象時會同時在其內部初始化保存一個Sync類型的對象到sync字段用于之后的同步操作。并且傳入Sync類構造器的count一定會大于等于0。 

public CountDownLatch(int count) {  
    if (count < 0) throw new IllegalArgumentException("count < 0");  
    this.sync = new Sync(count);  
} 

有了Sync類型的對象之后，我們在await()方法里就可以直接調用sync的acquireSharedInterruptibly方法來獲取同步器并陷入阻塞，等待count值變為0了。在AQS的acquireSharedInterruptibly方法中會在調用我們重寫的tryAcquireShared方法獲取失敗時進入阻塞狀態，直到CountDownLatch的count值變為0時才能成功獲取到同步器。 

public void await() throws InterruptedException {  
    // 調用sync對象的獲取方法來進入鎖等待  
    sync.acquireSharedInterruptibly(1);  
} 

而在CountDownLatch的另一個減少count值的重要方法countDown()中，我們同樣是通過調用sync上的方法來實現具體的同步功能。在這里，AQS的releaseShared(1)方法中同樣會調用我們在Sync類中重寫的tryReleaseShared方法來執行釋放操作，并在tryReleaseShared方法返回true時去喚醒等待隊列中的阻塞等待線程，讓它們在count值為0時能夠繼續執行。 

public void countDown() {  
    sync.releaseShared(1);  
} 

從上文中可以看出，CoundDownLatch中的各種功能都是通過內部類Sync來實現的，而這個Sync類就是一個繼承自AQS的子類。通過在內部類Sync中重寫了AQS的tryAcquireShared和tryReleaseShared兩個方法，我們就指定了AQS的行為策略，使其能夠符合我們對CountDownLatch功能的期望。這就是AQS的使用方法，下面我們來看一個大家可能會更熟悉的例子，來進一步了解AQS在獨占模式下的用法。

ReentrantLock的實現

可重入鎖ReentrantLock可以說是我們的老朋友了，從最早的synchronized關鍵字開始，我們就開始使用類似的功能了。可重入鎖的特點主要有兩點：

•  同一時間只能有一個線程持有
  
  •   如果我想保護一段代碼同一時間只能被一個線程所訪問，比如對一個隊列的插入操作。那么如果有一個線程已經獲取了鎖之后在修改隊列了，那么其他也想要修改隊列的線程就會陷入阻塞，等待之前的這個線程執行完成。

•  同一線程可以對一個鎖重復獲取成功多次
  •   而如果一個線程對同一個隊列執行了兩個插入操作，那么第二次獲取鎖時仍然會成功，而不會被***次成功獲取到的鎖所阻塞。

ReentrantLock類的常用操作主要有三種：

•  獲取鎖，一個線程一旦獲取鎖成功后就會阻塞其他線程獲取同一個鎖的操作，所以一旦獲取失敗，那么當前線程就會被阻塞
  
  •   最簡單的獲取鎖方法就是調用public void lock()方法
•  釋放鎖，獲取鎖之后就要在使用完之后釋放它，否則別的線程都將會因無法獲取鎖而被阻塞，所以我們一般會在finally中進行鎖的釋放操作
  •   可以通過調用ReentrantLock對象的unlock方法來釋放鎖
•  獲取條件變量，條件變量是和互斥鎖搭配使用的一種非常有用的數據結構，有興趣的讀者可以通過《從0到1實現自己的阻塞隊列(上)》這篇文章來了解條件變量具體的使用方法
  •   我們可以通過Condition newCondition()方法來獲取條件變量對象，然后調用條件變量對象上的await()、signal()、signalAll()方法來進行使用

ReentrantLock的內部Sync類

在ReentrantLock類中存在兩種AQS的子類，一個實現了非公平鎖，一個實現了公平鎖。所謂的“公平”指的就是獲取互斥鎖成功返回的時間會和獲取鎖操作發起的時間順序一致，例如有線程A已經持有了互斥鎖，當線程B、C、D按字母順序獲取鎖并進入等待，線程A釋放鎖后一定是線程B被喚醒，線程B釋放鎖后一定是C先被喚醒。也就是說鎖被釋放后對等待線程的喚醒順序和獲取鎖操作的順序一致。而且如果在這個過程中，有其他線程發起了獲取鎖操作，因為等待隊列中已經有線程在等待了，那么這個線程一定要排到等待隊列***去，而不能直接搶占剛剛被釋放還未被剛剛被喚醒的線程鎖持有的鎖。

下面我們同樣先看一下ReentrantLock類中定義的AQS子類Sync的具體源代碼。下面是上一段說到的非公平Sync類和公平Sync類兩個類的共同父類Sync的帶注釋源代碼，里面包含了大部分核心功能的實現。雖然下面包含了該類完整的源代碼，但是我們現在只需要關心三個核心操作，也是我們在獨占模式下需要重寫的三個AQS方法：tryAcquire、tryRelease和isHeldExclusively。建議在看完文章之后再回來回顧該類中其他的方法實現，直接跳過其他的方法當然也是完全沒有問題的。 

abstract static class Sync extends AbstractQueuedSynchronizer {  
    private static final long serialVersionUID = -5179523762034025860L;  
    /**  
     * 實現Lock接口的lock方法，子類化的主要原因是為了非公平版本的快速實現    
      */  
    abstract void lock();  
    /**  
     * 執行非公平的tryLock。tryAcquire方法在子類中被實現，但是兩者都需要非公平版本的trylock方法實現。  
     */  
    final boolean nonfairTryAcquire(int acquires) {  
        final Thread current = Thread.currentThread();  
        int c = getState();  
        // 如果鎖還未被持有  
        if (c == 0) {  
            // 通過CAS嘗試獲取鎖  
            if (compareAndSetState(0, acquires)) {  
                // 如果鎖獲取成功則將鎖持有者改為當前線程，并返回true  
                setExclusiveOwnerThread(current);  
                return true;  
            }  
        }  
        // 鎖已經被持有，則判斷鎖的持有者是否是當前線程  
        else if (current == getExclusiveOwnerThread()) {  
            // 可重入鎖，如果鎖的持有者是當前線程，那就在state上加上新的獲取數  
            int nextc = c + acquires;  
            // 判斷新的state值有沒有溢出  
            if (nextc < 0) // overflow  
                throw new Error("Maximum lock count exceeded");  
            // 將新的state更新為新的值，因為可以進入這段代碼的只有一個線程  
            // 所以不需要線程安全措施  
            setState(nextc);  
            return true;  
        }      
         return false;  
    }  
    // 重寫了AQS的獨占式釋放鎖方法  
    protected final boolean tryRelease(int releases) {  
        // 計算剩余的鎖持有量  
        // 因為只有當前線程持有該鎖的情況下才能執行這個方法，所以不需要做多線程保護  
        int c = getState() - releases;  
        // 如果當前線程未持有鎖，則直接拋出錯誤  
        if (Thread.currentThread() != getExclusiveOwnerThread())  
            throw new IllegalMonitorStateException();  
        boolean free = false;  
        // 如果鎖持有數已經減少到0，則釋放該鎖，并清空鎖持有者  
        if (c == 0) {  
            free = true;  
            setExclusiveOwnerThread(null);  
        }  
        // 更新state值，只有state值被設置為0才是真正地釋放了鎖  
        // 所以setState和setExclusiveOwnerThread之間不需要額外的同步措施  
        setState(c);  
        return free;  
    }   
    // 當前線程是否持有該鎖  
    protected final boolean isHeldExclusively() {  
        return getExclusiveOwnerThread() == Thread.currentThread();  
    }  
    // 創建對應的條件變量  
    final ConditionObject newCondition() {  
        return new ConditionObject();  
    }  
    // 從外層傳遞進來的方法  
    // 獲取當前的鎖持有者  
    final Thread getOwner() {  
        return getState() == 0 ? null : getExclusiveOwnerThread();  
    }  
    // 獲取鎖的持有計數  
    // 如果當前線程持有了該鎖則返回state值，否則返回0  
    final int getHoldCount() {  
        return isHeldExclusively() ? getState() : 0;  
    }  
    // 判斷鎖是否已經被持有  
    final boolean isLocked() {  
        return getState() != 0;  
    }  
} 

實際的tryAcquire方法將在公平Sync類與非公平Sync類兩個子類中實現，但是這兩個子類都需要調用父類Sync中的非公平版本的tryAcquire——nonfairTryAcquire方法。在這個方法中，我們主要做兩件事：

•  當前鎖還未被人持有。在ReentrantLock中使用AQS的state來保存鎖的狀態，state等于0時代表鎖沒有被任何線程持有，如果state大于0，那么就代表持有者對該鎖的重復獲取次數
  •   如果當前鎖還未被線程持有，那么就會通過compareAndSetState來原子性地修改state值，修改成功則需要設置當前線程為鎖的持有線程并返回true代表獲取成功；否則就返回
•  鎖已被當前線程持有
  •   在鎖已被當前線程持有的情況下，就需要將state值加1代表持有者線程對鎖的重復獲取次數。

而對于獨占式釋放同步器的tryRelease方法，則在父類Sync中直接實現了，兩個公平/非公平子類調用的都是同一段代碼。首先，只有鎖的持有者才能釋放鎖，所以如果當前線程不是所有者線程在釋放操作中就會拋出異常。如果釋放操作會將持有計數清零，那么當前線程就不再是該鎖的持有者了，鎖會被完全釋放，而鎖的所有者會被設置為null。***，Sync會將減掉入參中的釋放數之后的新持有計數更新到AQS的state中，并返回鎖是否已經被完全釋放了。

isHeldExclusively方法比較簡單，它只是檢查鎖的持有者是否是當前線程。

非公平Sync類的實現

Sync的兩個公平/非公平子類的實現比較簡單，下面是非公平版本子類的源代碼。在非公平版本的實現中，調用lock方法首先會嘗試通過CAS修改AQS的state值來直接搶占鎖，如果搶占成功就直接將持有者設置為當前線程；如果搶占失敗就調用acquire方法走正常流程來獲取鎖。而在acquire方法中就會調用子類中的tryAcquire方法并進一步調用到上文提到的父類中的nonfairTryAcquire方法來完成鎖獲取操作。 

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);  
    }  
    // 實現AQS方法，使用nonfairTryAcquire實現  
    protected final boolean tryAcquire(int acquires) {  
        return nonfairTryAcquire(acquires);  
    }  
} 

公平Sync類的實現

而在公平版本的Sync子類FairSync中，為了保證成功獲取到鎖的順序一定要和發起獲取鎖操作的順序一致，所以自然不能在lock方法中進行CAS方式的搶占，只能老老實實調用acquire方法走正式流程。而acquire方法最終就會調用子類中定義的tryAcquire來真正獲取鎖。

在tryAcquire方法中，代碼主要處理了兩種情況：

•  當前鎖還沒有被線程鎖持有
  •   只有在確保等待隊列為空的情況下才能嘗試用CAS方式直接搶占鎖，而在等待隊列不為空的情況下，***返回了false，之后acquire方法中的代碼會將當前線程放入到等待隊列中阻塞等待鎖的釋放。這就保證了在獲取鎖時已經有線程等待的情況下，任何線程都要進入等待隊列去等待獲取鎖，而不能直接對鎖進行獲取。
•  當前線程已經持有了該鎖
  •   如果當前線程已經是該鎖的持有者了，那么就會在state值上加上本次的獲取數量來更新鎖的重復獲取次數，并返回true代表獲取鎖成功。 

static final class FairSync extends Sync {  
    private static final long serialVersionUID = -3000897897090466540L;  
    // 直接使用acquire進行獲取鎖操作  
    final void lock() {  
        acquire(1);  
    }  
    /**  
     * 公平版本的tryAcquire方法。不要授予訪問權限，除非是遞歸調用或者沒有等待線程或者這是***個調用  
     */  
    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()) {  
            // 計算新的state值  
            int nextc = c + acquires;  
            // 如果鎖計數溢出，則拋出異常  
            if (nextc < 0)  
                throw new Error("Maximum lock count exceeded");  
            // 設置state狀態值  
            setState(nextc);  
            return true;  
        }  
        return false;  
    }  
} 

ReentrantLock對Sync類對象的使用

***，我們來看看ReentrantLock類中的lock()、unlock()、newCondition方法對Sync類對象的使用方式。

首先是在構造器中，根據入參指定的公平/非公平模式創建不同的內部Sync類對象，如果是公平模式就是用FairSync類，如果是非公平模式就是用NonfairSync類。 

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

然后在互斥鎖的鎖定方法lock()中，ReentrantLock直接使用Sync類中的lock方法來實現了鎖的獲取功能。 

public void lock() {  
    // 調用sync對象的lock方法實現  
    sync.lock();  
} 

在unlock()方法中也是一樣的情況，ReentrantLock直接依賴Sync類對象來實現這個功能。 

public void unlock() {  
    // 調用了sync對象的release方法實現  
    sync.release(1);  
} 

***一個創建條件變量的方法則直接依賴于AQS中定義的方法，我們在ReentranctLock的Sync類中并不需要做任務額外的工作，AQS就能為我們做好所有的事情。 

public Condition newCondition() {  
    // 調用了sync對象繼承自AQS的`newCondition`方法實現  
    return sync.newCondition();  
} 

通過ReentrantLock的例子我們能夠更明顯地感受到，這些基于AQS實現同步功能的類中并不需要做太多額外的工作，大多數操作都是通過直接調用Sync類對象上的方法來實現的。只要定義好了繼承自AQS的子類Sync，并通過Sync類重寫幾個AQS的關鍵方法來指定AQS的行為策略，就可以實現風格迥異的各種同步工具類了。

總結

在這篇文章中，我們從AQS的基本概念說起，簡單介紹了AQS的具體用法，然后通過CountDownLatch和ReentrantLock兩個常用的多線程同步工具類的源碼來具體了解了AQS的使用方式。我們不僅可以完全弄明白這兩個線程同步類的實現原理與細節，而且最重要的是找到了AQS這個幕后大BOSS。通過AQS，我們不僅可以更容易地閱讀并理解其他同步工具類的使用與實現，而且甚至可以動手開發出我們自己的自定義同步工具類。

到了這里，這一系列多線程編程相關的技術文章就接近尾聲了。后續我還會發布一篇囊括這個系列所有內容的總結性文章，里面會對多線程編程相關的知識脈絡做一次全面的梳理，然后將每個知識點鏈接到具體闡釋這個主題的文章中去。讓讀者可以在宏觀和微觀兩個層面理解多線程編程的原理與技巧，幫助大家建立完整的Java多線程理論與實踐知識體系。有興趣的讀者可以關注一下后續的文章，感謝大家的支持。