淺析 ReentrantLock

Java · 發表 2019-04-01 11:23:04

摘要：下面將從以下幾個方面淺析ReentrantLock： ReetrantLock可重入鎖簡介 ReetrantLock的特性中斷響應鎖申請等待限時 ReentrantLock中的公平鎖與非公平鎖 ReetrantLoc...

下面將從以下幾個方面淺析ReentrantLock：

ReetrantLock可重入鎖簡介
ReetrantLock的特性
- 中斷響應
- 鎖申請等待限時
ReentrantLock中的公平鎖與非公平鎖
ReetrantLock的內部實現

可重入鎖簡介

重入鎖 ReentrantLock，顧名思義，就是支援重進入的鎖，它表示該鎖能夠支援一個執行緒對資源的重複加鎖。程式碼示例如下：

public class ReenterLock implements Runnable{
private ReentrantLock lock = new ReentrantLock();
private int i = 0;
@Override
public void run() {
for (int j = 0; j < 1000000 ; j++) {
//獲取鎖
lock.lock();
try{
i++;
} finally {
//釋放鎖
lock.unlock();
}
}
}

public static void main(String[] args) throws InterruptedException {
ReenterLock reenterLock = new ReenterLock();
Thread t1 = new Thread(reenterLock);
Thread t2 = new Thread(reenterLock);
t1.start();
t2.start();
t1.join();
t2.join();
System.out.println(reenterLock.i);
}
}
複製程式碼

重進入是指任意執行緒在獲取到鎖之後能夠再次獲取該鎖而不會被鎖所阻塞：

執行緒的再次獲取鎖 。鎖需要去識別 獲取鎖的執行緒是否為當前佔據鎖的執行緒 ，如果是，則再次成功獲取
鎖的最終釋放 。執行緒重複n次獲取了鎖，隨後在第 n 次釋放該鎖，其他執行緒能夠獲取到該鎖。鎖的最終釋放要求鎖對於獲取進行計數自增，計數表示當前鎖被重複獲取的次數，而鎖被釋放，計數自減，當計數等於0時表示鎖已經成功釋放。

程式碼示例如下：

lock.lock();
lock.lock();
try{
i++;
} finally {
//釋放鎖
lock.unlock();
lock.unlock();
}
複製程式碼

ReentrantLock 處理死鎖的手段

ReentrantLock處理死鎖的手段，說白了也是ReentrantLock的重要特性

首先介紹下死鎖的大致概念：

兩個或多個程序在執行過程中，因爭奪資源而造成的一種相互等待的現象，如無外力作用，它們將無法繼續進行下去

下面舉一個 Synchronized下的死鎖例子：

public class DeadLockExample implements Runnable{
private boolean flag;
//鎖1
private static Object lock1 = new Object();
//鎖2
private static Object lock2 = new Object();

public DeadLockExample(boolean flag) {
this.flag = flag;
}

@Override
public void run() {
if (flag) {
synchronized (lock1) {
System.out.println("執行緒 ： " + Thread.currentThread().getName() + " get lock1");
try {
TimeUnit.SECONDS.sleep(2);
} catch (InterruptedException e) {
e.printStackTrace();
}
//嘗試獲取lock2
System.out.println("執行緒 :"+ Thread.currentThread().getName()+" waiting get lock2");
synchronized (lock2) {
System.out.println("執行緒 ： " + Thread.currentThread().getName() + " get lock1");
}
}
} else {
synchronized (lock2) {
System.out.println("執行緒 ： " + Thread.currentThread().getName() + " get lock2");
try {
TimeUnit.SECONDS.sleep(2);
} catch (InterruptedException e) {
e.printStackTrace();
}
//嘗試獲取鎖1
System.out.println("執行緒 :"+ Thread.currentThread().getName()+" waiting get lock1");
synchronized (lock1) {
System.out.println("執行緒 ： " + Thread.currentThread().getName() + " get lock1");
}
}
}
}

public static void main(String[] args) {
Thread t1 = new Thread(new DeadLockExample(true));
t1.setName("A");
Thread t2 = new Thread(new DeadLockExample(false));
t2.setName("B");
t1.start();
t2.start();
}
}
複製程式碼

輸出結果：

執行緒 ： A get lock1
執行緒 ： B get lock2
執行緒 :A waiting get lock2
執行緒 :B waiting get lock1
複製程式碼

可以看出執行緒 A在等待獲取鎖2，而執行緒 B在等待獲取鎖1，兩個執行緒相互等待這樣就形成了死鎖

而ReentranLock 與 Synchronized 一樣是一種同步機制，但是 ReentranLock 提供了比 synchronized 更強大、更靈活的鎖機制， 可以減少死鎖發生的概率 。

ReentranLock 提供了兩種方式來處理死鎖：

中斷響應
鎖申請等待限時

中斷響應

使用 lock的 lockInteruptibly() 方法獲取鎖，如果出現死鎖的話，呼叫執行緒的 interrupt來消除死鎖，以上面那個例子為基礎，改成 ReentrantLock的形式，程式碼如下

public class DeadLockWithReentrantLock implements Runnable{
private boolean flag;
//鎖1
private static ReentrantLock lock1 = new ReentrantLock();
//鎖2
private static ReentrantLock lock2 = new ReentrantLock();

public DeadLockWithReentrantLock(boolean flag) {
this.flag = flag;
}

@Override
public void run() {
try {
if (flag) {
//獲取鎖
lock1.lockInterruptibly();
System.out.println("執行緒 ： " + Thread.currentThread().getName() + " get lock1");
TimeUnit.SECONDS.sleep(2);
System.out.println("執行緒 ： " + Thread.currentThread().getName() + " try to get lock2");
lock2.lockInterruptibly();
} else {
lock2.lockInterruptibly();
System.out.println("執行緒 ： " + Thread.currentThread().getName() + " get lock2");
TimeUnit.SECONDS.sleep(2);
System.out.println("執行緒 ： " + Thread.currentThread().getName() + " try to get lock1");
lock1.lockInterruptibly();
}
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
//如果當前執行緒持有鎖1,釋放鎖1
if (lock1.isHeldByCurrentThread()) {
lock1.unlock();
}
//如果當前執行緒持有鎖2,釋放鎖2
if (lock2.isHeldByCurrentThread()) {
lock2.unlock();
}
System.out.println("執行緒 ： " + Thread.currentThread().getName() + " 退出");
}
}

public static void main(String[] args) throws InterruptedException {
Thread t1 = new Thread(new DeadLockWithReentrantLock(true));
t1.setName("A");
Thread t2 = new Thread(new DeadLockWithReentrantLock(false));
t2.setName("B");
t1.start();
t2.start();
TimeUnit.SECONDS.sleep(5);
System.out.println("執行緒B設定中斷標記，執行緒B將退出死鎖狀態");
t2.interrupt();

}
}
複製程式碼

輸出結果

執行緒 ： A get lock1
執行緒 ： B get lock2
執行緒 ： B try to get lock1
執行緒 ： A try to get lock2
執行緒B設定中斷標記，執行緒B將退出死鎖狀態
java.lang.InterruptedException
執行緒 ： B 退出
執行緒 ： A 退出
at java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireInterruptibly(AbstractQueuedSynchronizer.java:898)
at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireInterruptibly(AbstractQueuedSynchronizer.java:1222)
at java.util.concurrent.locks.ReentrantLock.lockInterruptibly(ReentrantLock.java:335)
at com.pjmike.thread.reentrantlock.DeadLockWithReentrantLock.run(DeadLockWithReentrantLock.java:36)
at java.lang.Thread.run(Thread.java:745)
複製程式碼

執行緒A獲取鎖1，執行緒B獲取鎖2，執行緒A嘗試獲取鎖2，執行緒B嘗試獲取鎖1，兩個執行緒相互等待對方持有的鎖，故形成了死鎖。此時 main函式中，呼叫執行緒B的 interrupt 中斷執行緒，執行緒B響應中斷，最後兩個執行緒都相繼退出。真正完成任務只有執行緒A,執行緒B首先響應中斷，放棄任務直接退出，釋放資源。

下面來看下關鍵方法 lockInterruptibly 是如何實現的：

public void lockInterruptibly() throws InterruptedException {
sync.acquireInterruptibly(1);
}
複製程式碼

方法中呼叫佇列同步器 AbstractQueuedSynchronizer 中的 acquireInterruptibly 方法

public final void acquireInterruptibly(int arg)
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
if (!tryAcquire(arg))
doAcquireInterruptibly(arg);
}
複製程式碼

從上面的程式碼就可以看出如果當前執行緒被中斷，就會丟擲一個 InterruptedException 異常，我們之前的輸出結果也是丟擲一箇中斷異常，最終死鎖被消除。關於佇列同步器的部分，這裡就不詳細介紹了，可以參閱《Java併發程式設計的藝術》一書，書中對AQS的描述如下:

AQS 是用來構建鎖或者其他同步元件的基礎框架，它使用了一個 int成員變量表示同步狀態，通過內建的FIFO 佇列來完成資源獲取執行緒的排隊工作。

鎖申請等待限時

除了等待外部中斷外，避免死鎖還有一種方法就是限時等待。限時等待的方式是呼叫 tryLock 方法，還是先來看程式碼示例如下：

public class DeadLockWithReentrantLock2 implements Runnable{
private boolean flag;
//鎖1
private static ReentrantLock lock1 = new ReentrantLock();
//鎖2
private static ReentrantLock lock2 = new ReentrantLock();

public DeadLockWithReentrantLock2(boolean flag) {
this.flag = flag;
}

@Override
public void run() {
try {
if (flag) {
if (lock1.tryLock()) {
System.out.println("執行緒 ： " + Thread.currentThread().getName() + " get lock1");
TimeUnit.SECONDS.sleep(2);
System.out.println("執行緒 ： " + Thread.currentThread().getName() + " try to get lock2");
if (lock2.tryLock()) {
System.out.println("執行緒 ： " + Thread.currentThread().getName() + " already get lock2");
}
}
} else {
if (lock2.tryLock()) {
System.out.println("執行緒 ： " + Thread.currentThread().getName() + " get lock2");
TimeUnit.SECONDS.sleep(2);
System.out.println("執行緒 ： " + Thread.currentThread().getName() + " try to get lock1");
if (lock1.tryLock()) {
System.out.println("執行緒 ： " + Thread.currentThread().getName() + " already get lock1");
}
}
}
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
//如果當前執行緒持有鎖1,釋放鎖1
if (lock1.isHeldByCurrentThread()) {
lock1.unlock();
}
//如果當前執行緒持有鎖2,釋放鎖2
if (lock2.isHeldByCurrentThread()) {
lock2.unlock();
}
System.out.println("執行緒 ： " + Thread.currentThread().getName() + " 退出");
}
}

public static void main(String[] args) throws InterruptedException {
Thread t1 = new Thread(new DeadLockWithReentrantLock2(true));
t1.setName("A");
Thread t2 = new Thread(new DeadLockWithReentrantLock2(false));
t2.setName("B");
t1.start();
t2.start();
TimeUnit.SECONDS.sleep(5);
}
}
複製程式碼

輸出結果是：

執行緒 ： B get lock2
執行緒 ： A get lock1
執行緒 ： B try to get lock1
執行緒 ： A try to get lock2
執行緒 ： B 退出
執行緒 ： A already get lock2
執行緒 ： A 退出
複製程式碼

ReentrantLock.tryLock()方法不帶引數執行的情況下，當前執行緒會嘗試獲取鎖，如果鎖並未被其他執行緒佔用，則申請鎖會成功，並立即返回true。如果鎖被其他執行緒佔用，則當前執行緒不會進行等待，而是立即返回 false.這種模式不會引起執行緒等待，因此也不會產生死鎖。

上面的例子中，執行緒A獲得鎖1，執行緒B獲得鎖2，執行緒B嘗試獲取鎖1，發現鎖1被佔用，此時執行緒B不會等待，最終退出釋放鎖2，執行緒A就獲得鎖2繼續執行任務而後退出。

其實，tryLock方法還可以接受兩個引數，一個表示等待時長，另外一個表示計時單位。

public boolean tryLock(long timeout, TimeUnit unit)
複製程式碼

比如設定時長為5s，就表示執行緒在鎖請求中，最多等待5s，如果超過5s沒有獲得鎖，就會返回 false.如果成功獲得鎖，則返回true.

ReetrantLock中的公平鎖與非公平鎖

ReentrantLock中有兩種鎖：公平鎖和非公平鎖。

公平鎖：按照時間順序，先來先獲取鎖，也就是FIFO，維護一個有序佇列
非公平鎖：請求獲取鎖的順序是隨機的，不是公平的，可能一個請求多次獲得鎖，一個請求一次鎖也獲得不了

預設情況下，ReentrantLock獲得的鎖是非公平的。上面舉的一些程式碼示例中獲得鎖都是非公平的。當然也可以設定公平鎖，在ReentrantLock的構造方法裡

public ReentrantLock(boolean fair)
複製程式碼

但是公平鎖需要系統維護一個有序佇列，因此公平鎖的實現成本比較高，效能也比較低下。下面來舉一個公平鎖的程式碼示例：

public class FairLock implements Runnable{
private static ReentrantLock lock = new ReentrantLock(true);
@Override
public void run() {
while (true) {
try {
lock.lock();
System.out.println(Thread.currentThread().getName() + " 獲得鎖 ");
} finally {
lock.unlock();
}
}
}

public static void main(String[] args) {
FairLock fairLock = new FairLock();
Thread A = new Thread(fairLock, "Thread-A");
Thread B = new Thread(fairLock, "Thread-B");
A.start();
B.start();
}
}
複製程式碼

輸出結果：

Thread-A 獲得鎖 
Thread-B 獲得鎖 
Thread-A 獲得鎖 
Thread-B 獲得鎖 
Thread-A 獲得鎖 
Thread-B 獲得鎖 
Thread-A 獲得鎖 
Thread-B 獲得鎖 
......
複製程式碼

從輸出結果看，兩個執行緒基本上是交替獲得鎖的，幾乎不會發生同一執行緒連續多次獲得鎖的可能，從而保證了公平性。

再次總結下公平鎖與非公平鎖：

公平鎖保證了鎖的獲取按照FIFO原則，而代價是進行大量的執行緒切換
非公平鎖雖然可能造成執行緒"飢餓",但極少的執行緒切換，保證了其更大的吞吐量

可重入鎖的內部實現

ReentrantLock的類層次結構如下圖所示：

ReentrantLock實現了Lock介面，Lock介面定義了鎖獲取和釋放的基本操作：

public interface Lock {
//獲取鎖
void lock();
//可中斷地獲取鎖，在鎖的獲取時可以中斷當前執行緒
void lockInterruptibly() throws InterruptedException;
//非阻塞的獲取鎖，呼叫該方法後立刻返回，如果能夠獲取返回true，否則返回false
boolean tryLock();
//獲取鎖的超時設定
boolean tryLock(long time, TimeUnit unit) throws InterruptedException;
//釋放鎖
void unlock();
//獲取等待通知元件
Condition newCondition();
}
複製程式碼

從上圖還可以看出，ReentrantLock內部有三個內部類：Sync、NonfairSync、FairSync。Sync是一個抽象型別，它繼承了AbstractQueuedSynchronizer(簡稱AQS)，而NonfairSync和FairSync是Sync的繼承類，分別對應非公平鎖和公平鎖。AQS是佇列同步器，是用來構建鎖或者其他同步元件的基礎框架，實現了很多與鎖相關的功能。

AQS 簡介

AQS的主要使用方式是繼承，子類通過繼承AQS並實現它的抽象方法來管理同步狀態。而Sync也是繼承AQS，實現了它的tryRelease方法。

在抽象方法的實現過程中免不了要對同步狀態進行更改，這時就需要使用AQS提供的三個方法:

getState(): 獲取當前同步狀態
setState(int newState): 設定當前同步狀態
compareAndSetState(int expect，int update): 使用CAS設定當前狀態，該方法能夠保證狀態設定的原子性。（ CAS是一種用於在多執行緒環境下實現同步功能的機制，CAS操作包含三個運算元--記憶體位置、預期數值和新值。CAS的實現邏輯是將記憶體位置處的數值與預期數值相比較、若相等，則將記憶體位置處的值替換為新值，若不相等，則不做任何操作）

同步器依賴內部的同步佇列(一個FIFO雙向佇列，也叫做CLH同步佇列)來完成同步狀態的管理，當前執行緒獲取獲取同步狀態失敗時，AQS則會將當前執行緒以及等待狀態等資訊構造成一個節點(Node)並將其加入CLH同步佇列，同時會阻塞當前執行緒，當同步狀態釋放時，會把首節點中的執行緒喚醒，使其再次嘗試獲取同步狀態。

最後再簡單介紹AQS中的幾個方法以方便後面分析使用，（AQS是一門大學問，可以說在Java併發是非常核心的內容，本文只做簡單介紹，對於AQS更詳細內容請參閱相關書籍）：

boolean tryAcquire(int arg): 獨佔式獲取同步狀態，實現該方法需要查詢當前狀態並判斷同步狀態是否符合預期，然後再進行CAS設定同步狀態
boolean tryRelease(int arg): 獨佔式釋放同步狀態，等待獲取同步狀態的執行緒將有機會獲取同步狀態
boolean release(int arg): 釋放同步狀態，並將CLH同步佇列中第一個節點包含的執行緒喚醒
void acquire(int arg): 獲取同步狀態，如果當前執行緒獲取同步狀態成功，則由該方法返回，否則，將會進入同步佇列等待，該方法將會呼叫重寫的tryAcquire(int arg)方法。

下面通過原始碼的形式，以非公平鎖為例，簡要分析lock方法與unlock的內部實現。

非公平鎖下的lock方法淺析

以下面這個demo的核心程式碼來分析：

private ReentrantLock lock = new ReentrantLock();
private int i = 0;
@Override
public void run() {
//獲取鎖
lock.lock();
try {
i++;
} finally {
//釋放鎖
lock.unlock();
}
}
複製程式碼

預設情況下，ReentrantLock使用非公平鎖，也就是NonfairSync，上述程式碼中 lock.lock() 實際呼叫的是NonfairSync的lock方法，lock內部首先執行compareAndSetState 方法進行CAS操作，嘗試搶佔鎖，如果成功，就呼叫 setExclusiveOwnerThread 方法把當前執行緒設定在這個鎖上，表示搶佔成功。

static final class NonfairSync extends Sync {
...
final void lock() {
//呼叫AQS的compareAndSetState方法進行CAS操作
//當同步狀態為0時，獲取鎖，並設定狀態為1
if (compareAndSetState(0, 1))
setExclusiveOwnerThread(Thread.currentThread());
else
acquire(1);
}
...
}
複製程式碼

如果鎖被其他執行緒搶佔，即失敗，則呼叫acquire(1)方法, 該方法是AQS提供的模板方法，總體原理是先去搶佔鎖，如果沒有搶佔成功，就在CLH佇列中增加一個的當前執行緒的節點，表示等待後續搶佔。

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

進入acquire方法，先呼叫tryAcquire，實則呼叫的是NonfairSync中的實現，然後再次跳轉到nonfairTryAcquire方法上。

// 1
protected final boolean tryAcquire(int acquires) {
return nonfairTryAcquire(acquires);
}
// 2
final boolean nonfairTryAcquire(int acquires) {
//當前執行緒
final Thread current = Thread.currentThread();
int c = getState();
//比較當前同步狀態是否為0，如果是0，就去搶佔鎖
if (c == 0) {
if (compareAndSetState(0, acquires)) {
setExclusiveOwnerThread(current);
return true;
}
}
//如果不為0，就比較當前執行緒與佔用鎖的執行緒是不是同一個執行緒，如果是，就去增加狀態變數的值
//這就是可重入鎖之所以能可重入，就是因為同一個執行緒可以反覆使用它的鎖
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;
}
複製程式碼

如果tryAcquire返回false，就進入acquireQueued方法向CLH同步佇列增加一個當前執行緒的節點，等待搶佔，關於其中的細節，這裡點到為止，不細說了。

下圖是NonfairSync的lock方法的一個呼叫時序圖，與上面的分析相呼應：

非公平鎖的unlock方法淺析

unlock呼叫過程原始碼如下：

//1 ReentrantLock中的unlock
public void unlock() {
sync.release(1); //呼叫Sync的release方法，實則呼叫AQS中的release
}
//2 AQS中的release
public final boolean release(int arg) {
if (tryRelease(arg)) {
Node h = head;
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}

//3在release中呼叫 Sync實現的tryRelease方法
protected final boolean tryRelease(int releases) {
//getState()=1，前面獲取鎖時已經更新為1，而releases為1，=> c =0
int c = getState() - releases; 
if (Thread.currentThread() != getExclusiveOwnerThread())
throw new IllegalMonitorStateException();
boolean free = false;
//去除鎖的獨佔執行緒
if (c == 0) {
free = true;
setExclusiveOwnerThread(null);
}
//重新設定state = 0
setState(c);
//釋放鎖成功返回true
return free;
}
複製程式碼

呼叫ReentrantLock中的unlock，實則直接呼叫AQS的release操作
進入release方法，內部呼叫tryRelease方法（Sync類已重寫該方法），去除鎖的獨佔執行緒，也就是釋放鎖
tryRelease內部實現是首先獲取同步狀態，然後將狀態減1，這裡減一主要是考慮到可重入鎖可能自身會多次佔用鎖，只有當同步狀態變成0時，才表示完全釋放了鎖。
一旦tryRelease釋放鎖成功，將CLH同步佇列中第一個節點包含的執行緒喚醒。