1-3w 字,一文詳解死鎖!
作者 | 王磊
來源 | Java 中文社羣(ID:javacn666)
轉載請聯繫授權(微信 ID:GG_Stone
死鎖(Dead Lock)指的是兩個或兩個以上的運算單元(進程、線程或協程),都在等待對方停止執行,以取得系統資源,但是沒有一方提前退出,就稱爲死鎖。
- 死鎖演示
死鎖的形成分爲兩個方面,一個是使用內置鎖 synchronized 形成的死鎖,另一種是使用顯式鎖 Lock 實現的死鎖,接下來我們分別來看。
1.1 死鎖 synchronized 版
publicclass DeadLockExample {
public static void main(String[] args) {
Object lockA = new Object(); // 創建鎖 A
Object lockB = new Object(); // 創建鎖 B
// 創建線程 1
Thread t1 = new Thread(new Runnable() {
@Override
public void run() {
// 先獲取鎖 A
synchronized (lockA) {
System.out.println("線程 1:獲取到鎖 A!");
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
// 嘗試獲取鎖 B
System.out.println("線程 1:等待獲取 B...");
synchronized (lockB) {
System.out.println("線程 1:獲取到鎖 B!");
}
}
}
});
t1.start(); // 運行線程
// 創建線程 2
Thread t2 = new Thread(new Runnable() {
@Override
public void run() {
// 先獲取鎖 B
synchronized (lockB) {
System.out.println("線程 2:獲取到鎖 B!");
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
// 嘗試獲取鎖 A
System.out.println("線程 2:等待獲取 A...");
synchronized (lockA) {
System.out.println("線程 2:獲取到鎖 A!");
}
}
}
});
t2.start(); // 運行線程
}
}
以上程序的執行結果如下:
1.2 死鎖 Lock 版
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
publicclass DeadLockByReentrantLockExample {
public static void main(String[] args) {
Lock lockA = new ReentrantLock(); // 創建鎖 A
Lock lockB = new ReentrantLock(); // 創建鎖 B
// 創建線程 1
Thread t1 = new Thread(new Runnable() {
@Override
public void run() {
lockA.lock(); // 加鎖
System.out.println("線程 1:獲取到鎖 A!");
try {
Thread.sleep(1000);
System.out.println("線程 1:等待獲取 B...");
lockB.lock(); // 加鎖
try {
System.out.println("線程 1:獲取到鎖 B!");
} finally {
lockB.unlock(); // 釋放鎖
}
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lockA.unlock(); // 釋放鎖
}
}
});
t1.start(); // 運行線程
// 創建線程 2
Thread t2 = new Thread(new Runnable() {
@Override
public void run() {
lockB.lock(); // 加鎖
System.out.println("線程 2:獲取到鎖 B!");
try {
Thread.sleep(1000);
System.out.println("線程 2:等待獲取 A...");
lockA.lock(); // 加鎖
try {
System.out.println("線程 2:獲取到鎖 A!");
} finally {
lockA.unlock(); // 釋放鎖
}
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lockB.unlock(); // 釋放鎖
}
}
});
t2.start(); // 運行線程
}
}
以上程序的執行結果如下:
- 死鎖產生原因
通過以上示例,我們可以得出結論,要產生死鎖需要滿足以下 4 個條件:
-
互斥條件:指運算單元(進程、線程或協程)對所分配到的資源具有排它性,也就是說在一段時間內某個鎖資源只能被一個運算單元所佔用。
-
請求和保持條件:指運算單元已經保持至少一個資源,但又提出了新的資源請求,而該資源已被其它運算單元佔有,此時請求運算單元阻塞,但又對自己已獲得的其它資源保持不放。
-
不可剝奪條件:指運算單元已獲得的資源,在未使用完之前,不能被剝奪。
-
環路等待條件:指在發生死鎖時,必然存在運算單元和資源的環形鏈,即運算單元正在等待另一個運算單元佔用的資源,而對方又在等待自己佔用的資源,從而造成環路等待的情況。
只有以上 4 個條件同時滿足,纔會造成死鎖問題。
- 死鎖排查工具
如果程序出現死鎖問題,可通過以下 4 種方案中的任意一種進行分析和排查。
3.1 jstack
我們在使用 jstack 之前,先要通過 jps 得到運行程序的進程 ID,使用方法如下:
有了進程 ID(PID)之後,我們就可以使用 “jstack -l PID” 來發現死鎖問題了,如下圖所示:
PS:可以使用 jstack -help 查看更多命令使用說明。
3.2 jconsole
使用 jconsole 需要打開 JDK 的 bin 目錄,找到 jconsole 並雙擊打開,如下圖所示:
3.3 jvisualvm
jvisualvm 也在 JDK 的 bin 目錄中,同樣是雙擊打開:
3.4 jmc
jmc 是 Oracle Java Mission Control 的縮寫,是一個對 Java 程序進行管理、監控、概要分析和故障排查的工具套件。它也是在 JDK 的 bin 目錄中,同樣是雙擊啓動,如下圖所示:
- 死鎖解決方案
4.1 死鎖解決方案分析
接下來我們來分析一下,產生死鎖的 4 個條件,哪些是可以破壞的?哪些是不能被破壞的?
-
互斥條件:系統特性,不能被破壞。
-
請求和保持條件:可以被破壞。
-
不可剝奪條件:系統特性,不能被破壞。
-
環路等待條件:可以被破壞。
通過上述分析,我們可以得出結論,我們只能通過破壞請求和保持條件或者是環路等待條件,從而來解決死鎖的問題,那上線,我們就先從破壞 “環路等待條件” 開始來解決死鎖問題。
4.2 解決方案 1:順序鎖
所謂的順序鎖指的是通過有順序的獲取鎖,從而避免產生環路等待條件,從而解決死鎖問題的。
當我們沒有使用順序鎖時,程序的執行可能是這樣的:
此時我們只需要將線程 1 和線程 2 獲取鎖的順序進行統一,也就是線程 1 和線程 2 同時執行之後,都先獲取鎖 A,再獲取鎖 B,執行流程如下圖所示:
順序鎖的實現代碼如下所示:
publicclass SolveDeadLockExample {
public static void main(String[] args) {
Object lockA = new Object(); // 創建鎖 A
Object lockB = new Object(); // 創建鎖 B
// 創建線程 1
Thread t1 = new Thread(new Runnable() {
@Override
public void run() {
synchronized (lockA) {
System.out.println("線程 1:獲取到鎖 A!");
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("線程 1:等待獲取 B...");
synchronized (lockB) {
System.out.println("線程 1:獲取到鎖 B!");
}
}
}
});
t1.start(); // 運行線程
// 創建線程 2
Thread t2 = new Thread(new Runnable() {
@Override
public void run() {
synchronized (lockA) {
System.out.println("線程 2:獲取到鎖 A!");
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("線程 2:等待獲取B...");
synchronized (lockB) {
System.out.println("線程 2:獲取到鎖 B!");
}
}
}
});
t2.start(); // 運行線程
}
}
以上程序的執行結果如下:
4.3 解決方案 2:輪詢鎖
輪詢鎖是通過打破 “請求和保持條件” 來避免造成死鎖的,它的實現思路簡單來說就是通過輪詢來嘗試獲取鎖,如果有一個鎖獲取失敗,則釋放當前線程擁有的所有鎖,等待下一輪再嘗試獲取鎖。
輪詢鎖的實現需要使用到 ReentrantLock 的 tryLock 方法,具體實現代碼如下:
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
publicclass SolveDeadLockExample {
public static void main(String[] args) {
Lock lockA = new ReentrantLock(); // 創建鎖 A
Lock lockB = new ReentrantLock(); // 創建鎖 B
// 創建線程 1(使用輪詢鎖)
Thread t1 = new Thread(new Runnable() {
@Override
public void run() {
// 調用輪詢鎖
pollingLock(lockA, lockB);
}
});
t1.start(); // 運行線程
// 創建線程 2
Thread t2 = new Thread(new Runnable() {
@Override
public void run() {
lockB.lock(); // 加鎖
System.out.println("線程 2:獲取到鎖 B!");
try {
Thread.sleep(1000);
System.out.println("線程 2:等待獲取 A...");
lockA.lock(); // 加鎖
try {
System.out.println("線程 2:獲取到鎖 A!");
} finally {
lockA.unlock(); // 釋放鎖
}
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lockB.unlock(); // 釋放鎖
}
}
});
t2.start(); // 運行線程
}
/**
* 輪詢鎖
*/
public static void pollingLock(Lock lockA, Lock lockB) {
while (true) {
if (lockA.tryLock()) { // 嘗試獲取鎖
System.out.println("線程 1:獲取到鎖 A!");
try {
Thread.sleep(1000);
System.out.println("線程 1:等待獲取 B...");
if (lockB.tryLock()) { // 嘗試獲取鎖
try {
System.out.println("線程 1:獲取到鎖 B!");
} finally {
lockB.unlock(); // 釋放鎖
System.out.println("線程 1:釋放鎖 B.");
break;
}
}
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lockA.unlock(); // 釋放鎖
System.out.println("線程 1:釋放鎖 A.");
}
}
// 等待一秒再繼續執行
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
以上程序的執行結果如下:
4.4 輪詢鎖優化
使用輪詢鎖雖然可以解決死鎖的問題,但並不是完美無缺的,比如以下這些問題。
4.4.1 問題 1:死循環
以上簡易版的輪詢鎖,如果遇到有一個線程一直霸佔或者長時間霸佔鎖資源的情況,就會導致這個輪詢鎖進入死循環的狀態,它會嘗試一直獲取鎖資源,這樣就會造成新的問題,帶來不必要的性能開銷,具體示例如下。
反例
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
publicclass SolveDeadLockExample {
public static void main(String[] args) {
Lock lockA = new ReentrantLock(); // 創建鎖 A
Lock lockB = new ReentrantLock(); // 創建鎖 B
// 創建線程 1(使用輪詢鎖)
Thread t1 = new Thread(new Runnable() {
@Override
public void run() {
// 調用輪詢鎖
pollingLock(lockA, lockB);
}
});
t1.start(); // 運行線程
// 創建線程 2
Thread t2 = new Thread(new Runnable() {
@Override
public void run() {
lockB.lock(); // 加鎖
System.out.println("線程 2:獲取到鎖 B!");
try {
Thread.sleep(1000);
System.out.println("線程 2:等待獲取 A...");
lockA.lock(); // 加鎖
try {
System.out.println("線程 2:獲取到鎖 A!");
} finally {
lockA.unlock(); // 釋放鎖
}
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
// 如果此處代碼未執行,線程 2 一直未釋放鎖資源
// lockB.unlock();
}
}
});
t2.start(); // 運行線程
}
/**
* 輪詢鎖
*/
public static void pollingLock(Lock lockA, Lock lockB) {
while (true) {
if (lockA.tryLock()) { // 嘗試獲取鎖
System.out.println("線程 1:獲取到鎖 A!");
try {
Thread.sleep(1000);
System.out.println("線程 1:等待獲取 B...");
if (lockB.tryLock()) { // 嘗試獲取鎖
try {
System.out.println("線程 1:獲取到鎖 B!");
} finally {
lockB.unlock(); // 釋放鎖
System.out.println("線程 1:釋放鎖 B.");
break;
}
}
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lockA.unlock(); // 釋放鎖
System.out.println("線程 1:釋放鎖 A.");
}
}
// 等待一秒再繼續執行
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
以上代碼的執行結果如下:
優化版
針對以上死循環的情況,我們可以改進的思路有以下兩種:
-
添加最大次數限制:如果經過了 n 次嘗試獲取鎖之後,還未獲取到鎖,則認爲獲取鎖失敗,執行失敗策略之後終止輪詢(失敗策略可以是記錄日誌或其他操作);
-
添加最大時長限制:如果經過了 n 秒嘗試獲取鎖之後,還未獲取到鎖,則認爲獲取鎖失敗,執行失敗策略之後終止輪詢。
以上策略任選其一就可以解決死循環的問題,出於實現成本的考慮,我們可以採用輪詢最大次數的方式來改進輪詢鎖,具體實現代碼如下:
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
publicclass SolveDeadLockExample {
public static void main(String[] args) {
Lock lockA = new ReentrantLock(); // 創建鎖 A
Lock lockB = new ReentrantLock(); // 創建鎖 B
// 創建線程 1(使用輪詢鎖)
Thread t1 = new Thread(new Runnable() {
@Override
public void run() {
// 調用輪詢鎖
pollingLock(lockA, lockB, 3);
}
});
t1.start(); // 運行線程
// 創建線程 2
Thread t2 = new Thread(new Runnable() {
@Override
public void run() {
lockB.lock(); // 加鎖
System.out.println("線程 2:獲取到鎖 B!");
try {
Thread.sleep(1000);
System.out.println("線程 2:等待獲取 A...");
lockA.lock(); // 加鎖
try {
System.out.println("線程 2:獲取到鎖 A!");
} finally {
lockA.unlock(); // 釋放鎖
}
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
// 線程 2 忘記釋放鎖資源
// lockB.unlock(); // 釋放鎖
}
}
});
t2.start(); // 運行線程
}
/**
* 輪詢鎖
*
* maxCount:最大輪詢次數
*/
public static void pollingLock(Lock lockA, Lock lockB, int maxCount) {
// 輪詢次數計數器
int count = 0;
while (true) {
if (lockA.tryLock()) { // 嘗試獲取鎖
System.out.println("線程 1:獲取到鎖 A!");
try {
Thread.sleep(1000);
System.out.println("線程 1:等待獲取 B...");
if (lockB.tryLock()) { // 嘗試獲取鎖
try {
System.out.println("線程 1:獲取到鎖 B!");
} finally {
lockB.unlock(); // 釋放鎖
System.out.println("線程 1:釋放鎖 B.");
break;
}
}
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lockA.unlock(); // 釋放鎖
System.out.println("線程 1:釋放鎖 A.");
}
}
// 判斷是否已經超過最大次數限制
if (count++ > maxCount) {
// 終止循環
System.out.println("輪詢鎖獲取失敗,記錄日誌或執行其他失敗策略");
return;
}
// 等待一秒再繼續嘗試獲取鎖
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
以上代碼的執行結果如下:
4.4.2 問題 2:線程餓死
我們以上的輪詢鎖的輪詢等待時間是固定時間,如下代碼所示:
// 等待 1s 再嘗試獲取(輪詢)鎖
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
這樣在特殊情況下會造成線程餓死的問題,也就是輪詢鎖一直獲取不到鎖的問題,比如以下示例。
反例
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
publicclass SolveDeadLockExample {
public static void main(String[] args) {
Lock lockA = new ReentrantLock(); // 創建鎖 A
Lock lockB = new ReentrantLock(); // 創建鎖 B
// 創建線程 1(使用輪詢鎖)
Thread t1 = new Thread(new Runnable() {
@Override
public void run() {
// 調用輪詢鎖
pollingLock(lockA, lockB, 3);
}
});
t1.start(); // 運行線程
// 創建線程 2
Thread t2 = new Thread(new Runnable() {
@Override
public void run() {
while (true) {
lockB.lock(); // 加鎖
System.out.println("線程 2:獲取到鎖 B!");
try {
System.out.println("線程 2:等待獲取 A...");
lockA.lock(); // 加鎖
try {
System.out.println("線程 2:獲取到鎖 A!");
} finally {
lockA.unlock(); // 釋放鎖
}
} finally {
lockB.unlock(); // 釋放鎖
}
// 等待一秒之後繼續執行
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
});
t2.start(); // 運行線程
}
/**
* 輪詢鎖
*/
public static void pollingLock(Lock lockA, Lock lockB, int maxCount) {
// 循環次數計數器
int count = 0;
while (true) {
if (lockA.tryLock()) { // 嘗試獲取鎖
System.out.println("線程 1:獲取到鎖 A!");
try {
Thread.sleep(100); // 等待 0.1s(獲取鎖需要的時間)
System.out.println("線程 1:等待獲取 B...");
if (lockB.tryLock()) { // 嘗試獲取鎖
try {
System.out.println("線程 1:獲取到鎖 B!");
} finally {
lockB.unlock(); // 釋放鎖
System.out.println("線程 1:釋放鎖 B.");
break;
}
}
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lockA.unlock(); // 釋放鎖
System.out.println("線程 1:釋放鎖 A.");
}
}
// 判斷是否已經超過最大次數限制
if (count++ > maxCount) {
// 終止循環
System.out.println("輪詢鎖獲取失敗,記錄日誌或執行其他失敗策略");
return;
}
// 等待一秒再繼續嘗試獲取鎖
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
以上代碼的執行結果如下:
優化版
接下來,我們可以將輪詢鎖的固定等待時間,改進爲固定時間 + 隨機時間的方式,這樣就可以避免因爲獲取鎖的頻率一致,而造成輪詢鎖 “餓死” 的問題了,具體實現代碼如下:
import java.util.Random;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
publicclass SolveDeadLockExample {
privatestatic Random rdm = new Random();
public static void main(String[] args) {
Lock lockA = new ReentrantLock(); // 創建鎖 A
Lock lockB = new ReentrantLock(); // 創建鎖 B
// 創建線程 1(使用輪詢鎖)
Thread t1 = new Thread(new Runnable() {
@Override
public void run() {
// 調用輪詢鎖
pollingLock(lockA, lockB, 3);
}
});
t1.start(); // 運行線程
// 創建線程 2
Thread t2 = new Thread(new Runnable() {
@Override
public void run() {
while (true) {
lockB.lock(); // 加鎖
System.out.println("線程 2:獲取到鎖 B!");
try {
System.out.println("線程 2:等待獲取 A...");
lockA.lock(); // 加鎖
try {
System.out.println("線程 2:獲取到鎖 A!");
} finally {
lockA.unlock(); // 釋放鎖
}
} finally {
lockB.unlock(); // 釋放鎖
}
// 等待一秒之後繼續執行
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
});
t2.start(); // 運行線程
}
/**
* 輪詢鎖
*/
public static void pollingLock(Lock lockA, Lock lockB, int maxCount) {
// 循環次數計數器
int count = 0;
while (true) {
if (lockA.tryLock()) { // 嘗試獲取鎖
System.out.println("線程 1:獲取到鎖 A!");
try {
Thread.sleep(100); // 等待 0.1s(獲取鎖需要的時間)
System.out.println("線程 1:等待獲取 B...");
if (lockB.tryLock()) { // 嘗試獲取鎖
try {
System.out.println("線程 1:獲取到鎖 B!");
} finally {
lockB.unlock(); // 釋放鎖
System.out.println("線程 1:釋放鎖 B.");
break;
}
}
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lockA.unlock(); // 釋放鎖
System.out.println("線程 1:釋放鎖 A.");
}
}
// 判斷是否已經超過最大次數限制
if (count++ > maxCount) {
// 終止循環
System.out.println("輪詢鎖獲取失敗,記錄日誌或執行其他失敗策略");
return;
}
// 等待一定時間(固定時間 + 隨機時間)之後再繼續嘗試獲取鎖
try {
Thread.sleep(300 + rdm.nextInt(8) * 100); // 固定時間 + 隨機時間
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
以上代碼的執行結果如下:
- 總結
本文介紹了死鎖的概念,以及產生死鎖的 4 個條件,排查死鎖可以通過本文提供的 4 種工具中的任意一種來檢測,從易用性和性能方面來考慮,推薦使用 jconsole 或 jvisualvm,最後我們介紹了死鎖問題的兩種解決方案:順序鎖和輪詢鎖。
本文由 Readfog 進行 AMP 轉碼,版權歸原作者所有。
來源:https://mp.weixin.qq.com/s/L9clv1virmmi32jwL9ZOfw