當前位置：首頁 > 编程资源 > 编程问答 >内容正文

编程问答

深入分析AbstractQueuedSynchronizer独占锁的实现原理：ReentranLock

發布時間：2024/1/17 编程问答 35 豆豆

生活随笔收集整理的這篇文章主要介紹了深入分析AbstractQueuedSynchronizer独占锁的实现原理：ReentranLock 小編覺得挺不錯的,現在分享給大家,幫大家做個參考.

一、ReentranLock

　　相信我們都使用過ReentranLock，ReentranLock是Concurrent包下一個用于實現并發的工具類（ReentrantReadWriteLock、Semaphore、CountDownLatch等），它和Synchronized一樣都是獨占鎖，它們兩個鎖的比較如下：?
　　1. ReentrantLock實現了Lock接口，提供了與synchronized同樣的互斥性和可見性，也同樣提供了可重入性。?
　　2. synchronized存在一些功能限制：無法中斷一個正在等待獲取鎖的線程，無法獲取一個鎖時無限得等待下去。ReentrantLock更加靈活，能提供更好的活躍性和性能，可以中斷線程?
　　3. 內置鎖的釋放時自動的，而ReentrantLock的釋放必須在finally手動釋放?
　　4. 在大并發量的時候，ReentranLock的效率會比Synchronized好很多?
　　5. Lock可以進行可中斷的（lock.lockInterruptibly()）、可超時的（tryLock(long time, TimeUnit unit)）、非阻塞（tryLock()）的方式獲取鎖?
　　更多關于Lock和synchronized：Java并發編程：Lock

　　一個并發工具自然最基本的功能就是獲取鎖和釋放鎖，那么有沒有想過，ReentranLock是如何來實現并發的？既然ReentranLock可以中斷線程，所以內部自然不可能使用synchronized來實現。事實上，ReentranLock只是一個工具類，它內部的的實現都是通過一個AbstractQueuedSynchronizer（簡稱AQS）來實現的，AQS是整個Concurrent包中最核心的地方，其它的并發工具也都是使用AQS來實現的，因此，以下我們就通過ReentranLock來分析AQS是如何實現的！

二、AQS

　　站在使用者的角度，AQS的功能可以分為兩類：獨占功能和共享功能，它的所有子類中，要么實現并使用了它獨占功能的API，要么使用了共享鎖的功能，而不會同時使用兩套API，即便是它最有名的子類ReentrantReadWriteLock，也是通過兩個內部類：讀鎖和寫鎖，分別實現的兩套API來實現的，為什么這么做，后面我們再分析，到目前為止，我們只需要明白AQS在功能上有獨占控制和共享控制兩種功能即可?
　　AQS類中，有一個叫做state的成員變量，在ReentranLock他表示獲取鎖的線程數，假如state=0，表示還沒有現成獲取鎖；1表示已經有現成獲取了鎖；大于1表示重入的數量

三、ReentranLock的源碼

　　首先我們要對ReentranLock有一個大體的了解，ReentranLock分為公平鎖和非公平鎖，并且ReentranLock是AQS獨占功能的體現?
　　公平鎖：每個線程搶占鎖的順序為先后調用lock方法的順序依次獲取鎖，就像排隊一樣?
　　非公平鎖：表示獲取鎖的線程是不定順序的，誰運氣好，誰就獲取到鎖?
　　
　　
　　可以看到，兩個鎖都是繼承了一個叫做Sync的類，并且都分別有兩個方法lock和tryAcquire，那我們看看Sync這個類：?
　　
　　原來，Sync繼承自AQS，并且公平鎖和非公平鎖的兩個方法lock和tryAcquire都是重寫了Sync的方法，這也就驗證了ReentrantLock的實現原理就是AQS

　　到這里，我們已經有了基本的認識，那么我們就想想，公平鎖和非公平鎖該如何實現：?
　　有那么一個被volatile修飾的標志位叫做key（其實就是上面所說的AQS中的state），用來表示有沒有線程拿走了鎖，還需要一個線程安全的隊列，維護一堆被掛起的線程，以至于當鎖被歸還時，能通知到這些被掛起的線程，可以來競爭獲取鎖了。?
　　因此，公平鎖和非公平鎖唯一的區別就是獲取鎖的時候，是先直接去獲取鎖還是先進入隊列中等待

四、ReentranLock的加鎖

　　我們來看看ReentranLock是如何加鎖的：

公平鎖

　　
　　公平鎖調用lock時，會直接調用父類AQS的acquire方法，這里傳入1，很簡單，就是告知有一個線程要獲取鎖，這里是定死的；因此，相反，在釋放鎖的時候，也是傳入1?
　　?
　　在acquire中，首先調用tryAcquire，目的嘗試獲取鎖，如果獲取不到，就調用addWaiter創建一個waiter（當前線程）防止到隊列中，然后自身阻塞，那我們來看看如何嘗試獲取鎖？（注意：兩個鎖都重寫了AQS的tryAcquire方法）

protected final boolean tryAcquire(int acquires) {//首先得到獲取鎖的當前線程final Thread current = Thread.currentThread();//獲取當前stateint c = getState();//如果當前沒有線程獲取鎖if (c == 0) {//hasQueuedPredecessors表示當前隊列是否有線程在等待//表示沒有線程在等待，同時采用CAS更新state的狀態if (!hasQueuedPredecessors() &&compareAndSetState(0, acquires)) {//然后設置一個屬性exclusiveOwnerThread = current，記錄鎖被當前線程拿去setExclusiveOwnerThread(current);return true;}}//如果c != 0，說明已經有線程獲取鎖，并且getExclusiveOwnerThread == current,表示當前正在獲取鎖的就是當前鎖，所以這里是重入！else if (current == getExclusiveOwnerThread()) {//重入的話，讓狀態為state+1，表示多一次重入int nextc = c + acquires;//如果當前狀態<0，說明出現異常if (nextc < 0)throw new Error("Maximum lock count exceeded");//設置當前標志位setState(nextc);return true;}//如果鎖已經被獲取，并且又不是重入，所以返回false，表明獲取鎖失敗return false;}}

　　獲取鎖的邏輯上面說得很明白了，但是這里需要了解的是CAS操作和隊列的數據結構，這個下面在說，我們接著看，回到tryAcquire中?
　　?
　　如果獲取鎖成功，則不操作；如果獲取鎖失敗，則調用addWaiter并采取Node.EXECLUSIVE模式把當前線程放到隊列中去，mode是一個表示Node類型的字段，僅僅表示這個節點是獨占的，還是共享的

private Node addWaiter(Node mode) {//把當前線程按照Node.EXECLUSIVE模式包裝成1個NodeNode node = new Node(Thread.currentThread(), mode);//用pred表示隊列中的尾節點Node pred = tail;//如果尾節點不為空if (pred != null) {node.prev = pred;//通過CAS操作把node插入到列表的尾部，并把尾節點指向node如果失敗，說明有并發，此時調用enqif (compareAndSetTail(pred, node)) {pred.next = node;return node;}}//如果隊列為空，或者CAS失敗，進入enq中死循環，“自旋”方式修改。enq(node);return node;}

　　先看下AQS中隊列的內存結構，我們知道，隊列由Node類型的節點組成，其中至少有兩個變量，一個封裝線程，一個封裝節點類型。?
　　而實際上，它的內存結構是這樣的（第一次節點插入時，第一個節點是一個空節點，代表有一個線程已經獲取鎖，事實上，隊列的第一個節點就是代表持有鎖的節點）：?
　　

private Node enq(final Node node) {//進入死循環for (;;) {Node t = tail;//如果尾節點為null，說明隊列為空if (t == null) {//此時通過CAS增加一個頭結點(即上圖的黃色節點)，并且tail也指向頭結點，之后下一次循環if (compareAndSetHead(new Node()))tail = head;} else {//否則，把當前線程的node插入到尾節點的后面node.prev = t;if (compareAndSetTail(t, node)) {t.next = node;//并返回插入結點的前一個節點return t;}}}}

　　這就完成了線程節點的插入，還需要做一件事：將當前線程掛起！，這里在acquireQueued內通過parkAndCheckInterrupt將線程掛起

final boolean acquireQueued(final Node node, int arg) {boolean failed = true;try {boolean interrupted = false;for (;;) {final Node p = node.predecessor();//如果當前的節點是head說明他是隊列中第一個“有效的”節點，因此嘗試獲取，if (p == head && tryAcquire(arg)) {//成功后，將上圖中的黃色節點移除，Node1變成頭節點。setHead(node);p.next = null; // help GCfailed = false;//返回true表示已經插入到隊列中，且已經做好了掛起的準備return interrupted;}//否則，檢查前一個節點的狀態為，看當前獲取鎖失敗的線程是否需要掛起。如果需要，借助JUC包下的LockSopport類的靜態方法Park掛起當前線程。知道被喚醒。if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) interrupted = true;}} finally {if (failed) //如果有異常cancelAcquire(node);// 取消請求，對應到隊列操作，就是將當前節點從隊列中移除。}}

　　這塊代碼有幾點需要說明：

　　1. Node節點中，除了存儲當前線程，節點類型，隊列中前后元素的變量，還有一個叫waitStatus的變量，改變量用于描述節點的狀態，為什么需要這個狀態呢？?
?
　　原因是：AQS的隊列中，在有并發時，肯定會存取一定數量的節點，每個節點[G4] 代表了一個線程的狀態，有的線程可能“等不及”獲取鎖了，需要放棄競爭，退出隊列，有的線程在等待一些條件滿足，滿足后才恢復執行（這里的描述很像某個J.U.C包下的工具類，ReentrankLock的Condition，事實上，Condition同樣也是AQS的子類）等等，總之，各個線程有各個線程的狀態，但總需要一個變量來描述它，這個變量就叫waitStatus,它有四種狀態：?
　　節點取消?
　　節點等待觸發?
　　節點等待條件?
　　節點狀態需要向后傳播。?
　　只有當前節點的前一個節點為SIGNAL時，才能當前節點才能被掛起。

　　2. 對線程的掛起及喚醒操作是通過使用UNSAFE類調用JNI方法實現的。當然，還提供了掛起指定時間后喚醒的API，在后面我們會講到。?
　　?
　　（這一塊分析來自：http://www.infoq.com/cn/articles/jdk1.8-abstractqueuedsynchronizer#anch140431）

　　我們來理一理思路：?
　　1. 調用lock方法獲取鎖，而lock方法內值調用了AQS的acquire(1)?
　　2. 然后嘗試獲取鎖，如果當前state標志==0，表示還沒有線程獲取鎖，然后再判斷是否有隊列在等待獲取該鎖，如果沒有隊列，說明當前線程是第一個獲取該鎖的線程，然后修改標志位，并且用一個變量exclusiveOwnerThread來記錄當前線程獲取了鎖?
　　3. 如果是重入狀態，也修改state+1?
　　4. 如果鎖已被占取，獲取失敗?
　　5. 如果獲取失敗，則把當前線程包裝成一個Node，插入到隊列中，?
　　6. 否則，檢查前一個節點的狀態為，看當前獲取鎖失敗的線程是否需要掛起。如果需要，借助JUC包下的LockSopport類的靜態方法Park掛起當前線程。知道被喚醒。

非公平鎖

　　這里可以看到，非公平鎖，首先是直接去獲取鎖，如果有并發獲取失敗，調用AQS的acquire（1），然后acquire中調用非公平鎖的tryAcquire，進而調用nonfairTryAcquire

final boolean nonfairTryAcquire(int acquires) {final Thread current = Thread.currentThread();int c = getState();//如果當前沒有現成獲取鎖，直接獲取鎖，然后設置一個屬性exclusiveOwnerThread = current，記錄鎖被當前線程拿去，這里和公平所有細微的差別，公平所還要判斷hasQueuedPredecessors（）if (c == 0) {if (compareAndSetState(0, acquires)) {setExclusiveOwnerThread(current);return true;}}//如果是重入else if (current == getExclusiveOwnerThread()) {int nextc = c + acquires;if (nextc < 0) // overflowthrow new Error("Maximum lock count exceeded");setState(nextc);return true;}//如果當前鎖獲取失敗，返回falsereturn false;}

　　其它的都和公平鎖一樣了，如果到這里都獲取失敗了，就會插入到隊列中阻塞起來

總結公平鎖和非公平鎖

公平鎖獲取鎖時，會老老實實得走AQS的流程去獲取鎖

非公平鎖獲取鎖是，首先會搶占鎖，達到不排隊的目的，如果搶占失敗，只能老老實實排隊了

五、ReentrantLock的釋放鎖

　　從上面我們可以知道，當鎖已被占，獲取鎖的線程會一直在隊列中排隊（FIFO），那么我們想想，釋放的時候該怎么做？?
　　1. 首先鎖的狀態位要改變?
　　2. 隊列中的頭結點去獲取鎖

　　我們來看看代碼驗證一下：?
　　釋放鎖的時候調用unlock()，然后在方法中調用AQS的release方法?
　　?
　　?
　　在release方法中，首先調用tryRelease方法，由于繼承自AQS的Sync類重寫了tryRelease方法，所以此時執行的是Sync的tryRelease方法

protected final boolean tryRelease(int releases) {//這里傳入的releases是1，跟獲取鎖時傳入的1一致，更新state狀態int c = getState() - releases;//如果當前占領鎖的線程不是嘗試釋放鎖的線程，會拋出非法異常if (Thread.currentThread() != getExclusiveOwnerThread())throw new IllegalMonitorStateException();boolean free = false;//如果釋放成功，則修改獲取鎖的變量為null，但是因為是重入的關系，不是每次釋放鎖c都等于0，直到最后一次釋放鎖時，才通知AQS不需要再記錄哪個線程正在獲取鎖if (c == 0) {free = true;setExclusiveOwnerThread(null);}setState(c);return free;}

　　此時已經釋放了鎖，然后便通知隊列頭部的線程去獲取鎖?
　　?
　　尋找的順序是從隊列尾部開始往前去找的最前面的一個waitStatus小于0的節點，找到這個及節點后，利用LockSopport類將其喚醒，這個waitStatu前面說過了，不記得了到前面看看。?
　　

六、總結

　　在Concurrent包中，基本上并發工具都是使用了AQS作為核心，因此AQS也是并發編程中最重要的地方！我們從ReentrantLock出發，去探討了AQS的實現原理，其實并不難，AQS中采用了一個state的狀態位+一個FIFO的隊列的方式，記錄了鎖的獲取，釋放等，這個state不一定用來代指鎖，ReentrantLock用它來表示線程已經重復獲取該鎖的次數，Semaphore用它來表示剩余的許可數量，FutureTask用它來表示任務的狀態（尚未開始，正在運行，已完成以及以取消）。同時，在AQS中也看到了很多CAS的操作。AQS有兩個功能：獨占功能和共享功能，而ReentranLock就是AQS獨占功能的體現，而CountDownLatch則是共享功能的體現

總結

以上是生活随笔為你收集整理的深入分析AbstractQueuedSynchronizer独占锁的实现原理：ReentranLock的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。

上一篇： java并发编程之AbstractQue
下一篇： Tengine HTTPS原理解析、实践