以下是本篇文章的大綱

1 synchronized和lock

    1.1 synchronized的局限性
    1.2 Lock簡介

2 AQS

3 lock()與unlock()實現(xiàn)原理

    3.1 基礎(chǔ)知識
    3.2 內(nèi)部結(jié)構(gòu)
    3.3 NonfairSync
    3.3.1 lock()
    3.3.2 unlock()
    3.3.3 小結(jié)
    3.4 FairSync

4 超時機制

5 總結(jié)

1 synchronized和lock

1.1 synchronized的局限性

synchronized是java內(nèi)置的關(guān)鍵字，它提供了一種獨占的加鎖方式。synchronized的獲取和釋放鎖由JVM實現(xiàn)，用戶不需要顯示的釋放鎖，非常方便。然而synchronized也有一定的局限性，例如：

當(dāng)線程嘗試獲取鎖的時候，如果獲取不到鎖會一直阻塞。

如果獲取鎖的線程進入休眠或者阻塞，除非當(dāng)前線程異常，否則其他線程嘗試獲取鎖必須一直等待。

JDK1.5之后發(fā)布，加入了Doug Lea實現(xiàn)的concurrent包。包內(nèi)提供了Lock類，用來提供更多擴展的加鎖功能。Lock彌補了synchronized的局限，提供了更加細粒度的加鎖功能。

1.2 Lock簡介

Lock api如下

void lock();
void lockInterruptibly() throws InterruptedException;
boolean tryLock();
boolean tryLock(long time, TimeUnit unit) throws InterruptedException;
void unlock();
Condition newCondition();

其中最常用的就是lock和unlock操作了。因為使用lock時，需要手動的釋放鎖，所以需要使用try..catch來包住業(yè)務(wù)代碼，并且在finally中釋放鎖。典型使用如下

private Lock lock = new ReentrantLock();
public void test(){
 lock.lock();
 try{
 doSomeThing();
 }catch (Exception e){
 // ignored
 }finally {
 lock.unlock();
 }
}

2 AQS

AbstractQueuedSynchronizer簡稱AQS，是一個用于構(gòu)建鎖和同步容器的框架。事實上concurrent包內(nèi)許多類都是基于AQS構(gòu)建，例如ReentrantLock，Semaphore，CountDownLatch，ReentrantReadWriteLock，F(xiàn)utureTask等。AQS解決了在實現(xiàn)同步容器時設(shè)計的大量細節(jié)問題。

AQS使用一個FIFO的隊列表示排隊等待鎖的線程，隊列頭節(jié)點稱作“哨兵節(jié)點”或者“啞節(jié)點”，它不與任何線程關(guān)聯(lián)。其他的節(jié)點與等待線程關(guān)聯(lián)，每個節(jié)點維護一個等待狀態(tài)waitStatus。如圖

AQS中還有一個表示狀態(tài)的字段state，例如ReentrantLocky用它表示線程重入鎖的次數(shù)，Semaphore用它表示剩余的許可數(shù)量，F(xiàn)utureTask用它表示任務(wù)的狀態(tài)。對state變量值的更新都采用CAS操作保證更新操作的原子性。

AbstractQueuedSynchronizer繼承了AbstractOwnableSynchronizer，這個類只有一個變量：exclusiveOwnerThread，表示當(dāng)前占用該鎖的線程，并且提供了相應(yīng)的get，set方法。

理解AQS可以幫助我們更好的理解JCU包中的同步容器。

3 lock()與unlock()實現(xiàn)原理

3.1 基礎(chǔ)知識

ReentrantLock是Lock的默認實現(xiàn)之一。那么lock()和unlock()是怎么實現(xiàn)的呢？首先我們要弄清楚幾個概念

可重入鎖?？芍厝腈i是指同一個線程可以多次獲取同一把鎖。ReentrantLock和synchronized都是可重入鎖。

可中斷鎖?？芍袛噫i是指線程嘗試獲取鎖的過程中，是否可以響應(yīng)中斷。synchronized是不可中斷鎖，而ReentrantLock則提供了中斷功能。

公平鎖與非公平鎖。公平鎖是指多個線程同時嘗試獲取同一把鎖時，獲取鎖的順序按照線程達到的順序，而非公平鎖則允許線程“插隊”。synchronized是非公平鎖，而ReentrantLock的默認實現(xiàn)是非公平鎖，但是也可以設(shè)置為公平鎖。

CAS操作(CompareAndSwap)。CAS操作簡單的說就是比較并交換。CAS 操作包含三個操作數(shù) —— 內(nèi)存位置（V）、預(yù)期原值（A）和新值(B)。如果內(nèi)存位置的值與預(yù)期原值相匹配，那么處理器會自動將該位置值更新為新值。否則，處理器不做任何操作。無論哪種情況，它都會在 CAS 指令之前返回該位置的值。CAS 有效地說明了“我認為位置 V 應(yīng)該包含值 A；如果包含該值，則將 B 放到這個位置；否則，不要更改該位置，只告訴我這個位置現(xiàn)在的值即可?！?Java并發(fā)包(java.util.concurrent)中大量使用了CAS操作,涉及到并發(fā)的地方都調(diào)用了sun.misc.Unsafe類方法進行CAS操作。

3.2 內(nèi)部結(jié)構(gòu)

ReentrantLock提供了兩個構(gòu)造器，分別是

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

默認構(gòu)造器初始化為NonfairSync對象，即非公平鎖，而帶參數(shù)的構(gòu)造器可以指定使用公平鎖和非公平鎖。由lock()和unlock的源碼可以看到，它們只是分別調(diào)用了sync對象的lock()和release(1)方法。

Sync是ReentrantLock的內(nèi)部類，它的結(jié)構(gòu)如下

可以看到Sync擴展了AbstractQueuedSynchronizer。

3.3 NonfairSync

我們從源代碼出發(fā)，分析非公平鎖獲取鎖和釋放鎖的過程。

3.3.1 lock()

lock()源碼如下

final void lock() {
 if (compareAndSetState(0, 1))
 setExclusiveOwnerThread(Thread.currentThread());
 else
 acquire(1);
}

首先用一個CAS操作，判斷state是否是0（表示當(dāng)前鎖未被占用），如果是0則把它置為1，并且設(shè)置當(dāng)前線程為該鎖的獨占線程，表示獲取鎖成功。當(dāng)多個線程同時嘗試占用同一個鎖時，CAS操作只能保證一個線程操作成功，剩下的只能乖乖的去排隊啦。

“非公平”即體現(xiàn)在這里，如果占用鎖的線程剛釋放鎖，state置為0，而排隊等待鎖的線程還未喚醒時，新來的線程就直接搶占了該鎖，那么就“插隊”了。

若當(dāng)前有三個線程去競爭鎖，假設(shè)線程A的CAS操作成功了，拿到了鎖開開心心的返回了，那么線程B和C則設(shè)置state失敗，走到了else里面。我們往下看acquire。

acquire(arg)

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

代碼非常簡潔，但是背后的邏輯卻非常復(fù)雜，可見Doug Lea大神的編程功力。

1. 第一步。嘗試去獲取鎖。如果嘗試獲取鎖成功，方法直接返回。

tryAcquire(arg)

final boolean nonfairTryAcquire(int acquires) {
 //獲取當(dāng)前線程
 final Thread current = Thread.currentThread();
 //獲取state變量值
 int c = getState();
 if (c == 0) { //沒有線程占用鎖
 if (compareAndSetState(0, acquires)) {
 //占用鎖成功,設(shè)置獨占線程為當(dāng)前線程
 setExclusiveOwnerThread(current);
 return true;
 }
 } else if (current == getExclusiveOwnerThread()) { //當(dāng)前線程已經(jīng)占用該鎖
 int nextc = c + acquires;
 if (nextc < 0) // overflow
 throw new Error("Maximum lock count exceeded");
 // 更新state值為新的重入次數(shù)
 setState(nextc);
 return true;
 }
 //獲取鎖失敗
 return false;
}

非公平鎖tryAcquire的流程是：檢查state字段，若為0，表示鎖未被占用，那么嘗試占用，若不為0，檢查當(dāng)前鎖是否被自己占用，若被自己占用，則更新state字段，表示重入鎖的次數(shù)。如果以上兩點都沒有成功，則獲取鎖失敗，返回false。

2. 第二步，入隊。由于上文中提到線程A已經(jīng)占用了鎖，所以B和C執(zhí)行tryAcquire失敗，并且入等待隊列。如果線程A拿著鎖死死不放，那么B和C就會被掛起。

先看下入隊的過程。

先看addWaiter(Node.EXCLUSIVE)

/**
 * 將新節(jié)點和當(dāng)前線程關(guān)聯(lián)并且入隊列
 * @param mode 獨占/共享
 * @return 新節(jié)點
 */
private Node addWaiter(Node mode) {
 //初始化節(jié)點,設(shè)置關(guān)聯(lián)線程和模式(獨占 or 共享)
 Node node = new Node(Thread.currentThread(), mode);
 // 獲取尾節(jié)點引用
 Node pred = tail;
 // 尾節(jié)點不為空,說明隊列已經(jīng)初始化過
 if (pred != null) {
 node.prev = pred;
 // 設(shè)置新節(jié)點為尾節(jié)點
 if (compareAndSetTail(pred, node)) {
 pred.next = node;
 return node;
 }
 }
 // 尾節(jié)點為空,說明隊列還未初始化,需要初始化head節(jié)點并入隊新節(jié)點
 enq(node);
 return node;
}

B、C線程同時嘗試入隊列，由于隊列尚未初始化，tail==null，故至少會有一個線程會走到enq(node)。我們假設(shè)同時走到了enq(node)里。

/**
 * 初始化隊列并且入隊新節(jié)點
 */
private Node enq(final Node node) {
 //開始自旋
 for (;;) {
 Node t = tail;
 if (t == null) { // Must initialize
 // 如果tail為空,則新建一個head節(jié)點,并且tail指向head
 if (compareAndSetHead(new Node()))
 tail = head;
 } else {
 node.prev = t;
 // tail不為空,將新節(jié)點入隊
 if (compareAndSetTail(t, node)) {
 t.next = node;
 return t;
 }
 }
 }
}

這里體現(xiàn)了經(jīng)典的自旋+CAS組合來實現(xiàn)非阻塞的原子操作。由于compareAndSetHead的實現(xiàn)使用了unsafe類提供的CAS操作，所以只有一個線程會創(chuàng)建head節(jié)點成功。假設(shè)線程B成功，之后B、C開始第二輪循環(huán)，此時tail已經(jīng)不為空，兩個線程都走到else里面。假設(shè)B線程compareAndSetTail成功，那么B就可以返回了，C由于入隊失敗還需要第三輪循環(huán)。最終所有線程都可以成功入隊。

當(dāng)B、C入等待隊列后，此時AQS隊列如下：

3. 第三步，掛起。B和C相繼執(zhí)行acquireQueued(final Node node, int arg)。這個方法讓已經(jīng)入隊的線程嘗試獲取鎖，若失敗則會被掛起。

/**
 * 已經(jīng)入隊的線程嘗試獲取鎖
 */
final boolean acquireQueued(final Node node, int arg) {
 boolean failed = true; //標(biāo)記是否成功獲取鎖
 try {
 boolean interrupted = false; //標(biāo)記線程是否被中斷過
 for (;;) {
 final Node p = node.predecessor(); //獲取前驅(qū)節(jié)點
 //如果前驅(qū)是head,即該結(jié)點已成老二，那么便有資格去嘗試獲取鎖
 if (p == head && tryAcquire(arg)) {
 setHead(node); // 獲取成功,將當(dāng)前節(jié)點設(shè)置為head節(jié)點
 p.next = null; // 原h(huán)ead節(jié)點出隊,在某個時間點被GC回收
 failed = false; //獲取成功
 return interrupted; //返回是否被中斷過
 }
 // 判斷獲取失敗后是否可以掛起,若可以則掛起
 if (shouldParkAfterFailedAcquire(p, node) &&
  parkAndCheckInterrupt())
 // 線程若被中斷,設(shè)置interrupted為true
 interrupted = true;
 }
 } finally {
 if (failed)
 cancelAcquire(node);
 }
}

code里的注釋已經(jīng)很清晰的說明了acquireQueued的執(zhí)行流程。假設(shè)B和C在競爭鎖的過程中A一直持有鎖，那么它們的tryAcquire操作都會失敗，因此會走到第2個if語句中。我們再看下shouldParkAfterFailedAcquire和parkAndCheckInterrupt都做了哪些事吧。

/**
 * 判斷當(dāng)前線程獲取鎖失敗之后是否需要掛起.
 */
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
 //前驅(qū)節(jié)點的狀態(tài)
 int ws = pred.waitStatus;
 if (ws == Node.SIGNAL)
 // 前驅(qū)節(jié)點狀態(tài)為signal,返回true
 return true;
 // 前驅(qū)節(jié)點狀態(tài)為CANCELLED
 if (ws > 0) {
 // 從隊尾向前尋找第一個狀態(tài)不為CANCELLED的節(jié)點
 do {
 node.prev = pred = pred.prev;
 } while (pred.waitStatus > 0);
 pred.next = node;
 } else {
 // 將前驅(qū)節(jié)點的狀態(tài)設(shè)置為SIGNAL
 compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
 }
 return false;
} 
/**
 * 掛起當(dāng)前線程,返回線程中斷狀態(tài)并重置
 */
private final boolean parkAndCheckInterrupt() {
 LockSupport.park(this);
 return Thread.interrupted();
}

線程入隊后能夠掛起的前提是，它的前驅(qū)節(jié)點的狀態(tài)為SIGNAL，它的含義是“Hi，前面的兄弟，如果你獲取鎖并且出隊后，記得把我喚醒！”。所以shouldParkAfterFailedAcquire會先判斷當(dāng)前節(jié)點的前驅(qū)是否狀態(tài)符合要求，若符合則返回true，然后調(diào)用parkAndCheckInterrupt，將自己掛起。如果不符合，再看前驅(qū)節(jié)點是否>0(CANCELLED)，若是那么向前遍歷直到找到第一個符合要求的前驅(qū)，若不是則將前驅(qū)節(jié)點的狀態(tài)設(shè)置為SIGNAL。

整個流程中，如果前驅(qū)結(jié)點的狀態(tài)不是SIGNAL，那么自己就不能安心掛起，需要去找個安心的掛起點，同時可以再嘗試下看有沒有機會去嘗試競爭鎖。

最終隊列可能會如下圖所示

線程B和C都已經(jīng)入隊，并且都被掛起。當(dāng)線程A釋放鎖的時候，就會去喚醒線程B去獲取鎖啦。

3.3.2 unlock()

unlock相對于lock就簡單很多。源碼如下

public void unlock() {
 sync.release(1);
} 
public final boolean release(int arg) {
 if (tryRelease(arg)) {
 Node h = head;
 if (h != null && h.waitStatus != 0)
 unparkSuccessor(h);
 return true;
 }
 return false;
}

如果理解了加鎖的過程，那么解鎖看起來就容易多了。流程大致為先嘗試釋放鎖，若釋放成功，那么查看頭結(jié)點的狀態(tài)是否為SIGNAL，如果是則喚醒頭結(jié)點的下個節(jié)點關(guān)聯(lián)的線程，如果釋放失敗那么返回false表示解鎖失敗。這里我們也發(fā)現(xiàn)了，每次都只喚起頭結(jié)點的下一個節(jié)點關(guān)聯(lián)的線程。

最后我們再看下tryRelease的執(zhí)行過程

/**
 * 釋放當(dāng)前線程占用的鎖
 * @param releases
 * @return 是否釋放成功
 */
protected final boolean tryRelease(int releases) {
 // 計算釋放后state值
 int c = getState() - releases;
 // 如果不是當(dāng)前線程占用鎖,那么拋出異常
 if (Thread.currentThread() != getExclusiveOwnerThread())
 throw new IllegalMonitorStateException();
 boolean free = false;
 if (c == 0) {
 // 鎖被重入次數(shù)為0,表示釋放成功
 free = true;
 // 清空獨占線程
 setExclusiveOwnerThread(null);
 }
 // 更新state值
 setState(c);
 return free;
}

這里入?yún)?。tryRelease的過程為：當(dāng)前釋放鎖的線程若不持有鎖，則拋出異常。若持有鎖，計算釋放后的state值是否為0，若為0表示鎖已經(jīng)被成功釋放，并且則清空獨占線程，最后更新state值，返回free。

3.3.3 小結(jié)

用一張流程圖總結(jié)一下非公平鎖的獲取鎖的過程。   

3.4 FairSync

公平鎖和非公平鎖不同之處在于，公平鎖在獲取鎖的時候，不會先去檢查state狀態(tài)，而是直接執(zhí)行aqcuire(1)，這里不再贅述。   

4 超時機制

在ReetrantLock的tryLock(long timeout, TimeUnit unit) 提供了超時獲取鎖的功能。它的語義是在指定的時間內(nèi)如果獲取到鎖就返回true，獲取不到則返回false。這種機制避免了線程無限期的等待鎖釋放。那么超時的功能是怎么實現(xiàn)的呢？我們還是用非公平鎖為例來一探究竟。

public boolean tryLock(long timeout, TimeUnit unit)
 throws InterruptedException {
 return sync.tryAcquireNanos(1, unit.toNanos(timeout));
}

還是調(diào)用了內(nèi)部類里面的方法。我們繼續(xù)向前探究

 public final boolean tryAcquireNanos(int arg, long nanosTimeout)
 throws InterruptedException {
 if (Thread.interrupted())
 throw new InterruptedException();
 return tryAcquire(arg) ||
 doAcquireNanos(arg, nanosTimeout);
}

這里的語義是：如果線程被中斷了，那么直接拋出InterruptedException。如果未中斷，先嘗試獲取鎖，獲取成功就直接返回，獲取失敗則進入doAcquireNanos。tryAcquire我們已經(jīng)看過，這里重點看一下doAcquireNanos做了什么。

/**
 * 在有限的時間內(nèi)去競爭鎖
 * @return 是否獲取成功
 */
private boolean doAcquireNanos(int arg, long nanosTimeout)
 throws InterruptedException {
 // 起始時間
 long lastTime = System.nanoTime();
 // 線程入隊
 final Node node = addWaiter(Node.EXCLUSIVE);
 boolean failed = true;
 try {
 // 又是自旋!
 for (;;) {
 // 獲取前驅(qū)節(jié)點
 final Node p = node.predecessor();
 // 如果前驅(qū)是頭節(jié)點并且占用鎖成功,則將當(dāng)前節(jié)點變成頭結(jié)點
 if (p == head && tryAcquire(arg)) {
 setHead(node);
 p.next = null; // help GC
 failed = false;
 return true;
 }
 // 如果已經(jīng)超時,返回false
 if (nanosTimeout <= 0)
 return false;
 // 超時時間未到,且需要掛起
 if (shouldParkAfterFailedAcquire(p, node) &&
  nanosTimeout > spinForTimeoutThreshold)
 // 阻塞當(dāng)前線程直到超時時間到期
 LockSupport.parkNanos(this, nanosTimeout);
 long now = System.nanoTime();
 // 更新nanosTimeout
 nanosTimeout -= now - lastTime;
 lastTime = now;
 if (Thread.interrupted())
 //相應(yīng)中斷
 throw new InterruptedException();
 }
 } finally {
 if (failed)
 cancelAcquire(node);
 }
}

doAcquireNanos的流程簡述為：線程先入等待隊列，然后開始自旋，嘗試獲取鎖，獲取成功就返回，失敗則在隊列里找一個安全點把自己掛起直到超時時間過期。這里為什么還需要循環(huán)呢？因為當(dāng)前線程節(jié)點的前驅(qū)狀態(tài)可能不是SIGNAL，那么在當(dāng)前這一輪循環(huán)中線程不會被掛起，然后更新超時時間，開始新一輪的嘗試

5 總結(jié)

ReentrantLock的核心功能講解差不多落下帷幕，理解AQS，就很容易理解ReentrantLock的實現(xiàn)原理。文中慘雜著筆者的個人理解，如有不正之處，還望指正。

 以上就是本文的全部內(nèi)容，希望本文的內(nèi)容對大家的學(xué)習(xí)或者工作能帶來一定的幫助，同時也希望多多支持腳本之家！