當我們遇到死鎖之后，除了可以手動重啟程序解決之外，還可以考慮是使用順序鎖和輪詢鎖，這部分的內容可以參考我的上一篇文章，這里就不再贅述了。然而，輪詢鎖在使用的過程中，如果使用不當會帶來新的嚴重問題，所以本篇我們就來了解一下這些問題，以及相應的解決方案。
問題演示
當我們沒有使用輪詢鎖之前，可能會出現這樣的問題：
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
public class DeadLockByReentrantLock {
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 {
lockA.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(); // 運行線程
}
}
以上代碼的執行結果如下：
從上述結果可以看出，此時程序中出現了線程相互等待，並嘗試獲取對方（鎖）資源的情況，這就是典型的死鎖問題了。
簡易版輪詢鎖
當出現死鎖問題之后，我們就可以使用輪詢鎖來解決它了，它的實現思路是通過輪詢的方式來獲取多個鎖，如果中途有任意一個鎖獲取失敗，則執行回退操作，釋放當前線程擁有的所有鎖，等待下一次重新執行，這樣就可以避免多個線程同時擁有並霸占鎖資源了，從而直接解決了死鎖的問題，簡易版的輪詢鎖實現如下：
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
public class SolveDeadLockExample2 {
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(); // 運行線程
}
/**
* 輪詢鎖
*/
private 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();
}
}
}
}
以上代碼的執行結果如下：
從上述結果可以看出，當我們在程序中使用輪詢鎖之后就不會出現死鎖的問題了，但以上輪詢鎖也並不是完美無缺的，下面我們來看看這個輪詢鎖會有什么樣的問題？
問題1：死循環
以上簡易版的輪詢鎖，如果遇到有一個線程一直霸占或者長時間霸占鎖資源的情況，就會導致這個輪詢鎖進入死循環的狀態，它會嘗試一直獲取鎖資源，這樣就會造成新的問題，帶來不必要的性能開銷，具體示例如下。
反例
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
public class 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();
}
}
}
}
以上代碼的執行結果如下：
從上述結果可以看出，線程 1 輪詢鎖進入了死循環的狀態。
優化版
針對以上死循環的情況，我們可以改進的思路有以下兩種：
- 添加最大次數限制：如果經過了 n 次嘗試獲取鎖之后，還未獲取到鎖，則認為獲取鎖失敗，執行失敗策略之后終止輪詢（失敗策略可以是記錄日志或其他操作）；
- 添加最大時長限制：如果經過了 n 秒嘗試獲取鎖之后，還未獲取到鎖，則認為獲取鎖失敗，執行失敗策略之后終止輪詢。
以上策略任選其一就可以解決死循環的問題，出於實現成本的考慮，我們可以采用輪詢最大次數的方式來改進輪詢鎖，具體實現代碼如下：
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
public class 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();
}
}
}
}
以上代碼的執行結果如下：
從以上結果可以看出，當我們改進之后，輪詢鎖就不會出現死循環的問題了，它會嘗試一定次數之后終止執行。
問題2：線程餓死
我們以上的輪詢鎖的輪詢等待時間是固定時間，如下代碼所示：
// 等待 1s 再嘗試獲取（輪詢）鎖
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
這樣在特殊情況下會造成線程餓死的問題，也就是輪詢鎖一直獲取不到鎖的問題，比如以下示例。
反例
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
public class 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();
}
}
}
}
以上代碼的執行結果如下：
從上述結果可以看出，線程 1（輪詢鎖）一直未成功獲取到鎖，造成這種結果的原因是：線程 1 每次輪詢的等待時間為固定的 1s，而線程 2 也是相同的頻率，每 1s 獲取一次鎖，這樣就會導致線程 2 會一直先成功獲取到鎖，而線程 1 則會一直處於“餓死”的情況，執行流程如下圖所示：
優化版
接下來，我們可以將輪詢鎖的固定等待時間，改進為固定時間 + 隨機時間的方式，這樣就可以避免因為獲取鎖的頻率一致，而造成輪詢鎖“餓死”的問題了，具體實現代碼如下：
import java.util.Random;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
public class SolveDeadLockExample {
private static 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();
}
}
}
}
以上代碼的執行結果如下：
從上述結果可以看出，線程 1（輪詢鎖）加入隨機等待時間之后就不會出現線程餓死的問題了。
總結
本文我們介紹了輪詢鎖的用途，用於解決死鎖問題，但簡易版的輪詢鎖在某些情況下會造成死循環和線程餓死的問題，因此我們對輪詢鎖進行了優化，給輪詢鎖加入了最大輪詢次數，以及隨機輪詢等待時間，這樣就可以解決因為引入輪詢鎖而造成的新問題了，這樣就可以愉快的使用它來解決死鎖的問題了。
參考 & 鳴謝
《Java並發編程實戰》
並發原創文章推薦
- 線程的 4 種創建方法和使用詳解！
- Java中用戶線程和守護線程區別這么大？
- 深入理解線程池 ThreadPool
- 線程池的7種創建方式，強烈推薦你用它...
- 池化技術到達有多牛？看了線程和線程池的對比嚇我一跳！
- 並發中的線程同步與鎖
- synchronized 加鎖 this 和 class 的區別！
- volatile 和 synchronized 的區別
- 輕量級鎖一定比重量級鎖快嗎？
- 這樣終止線程，竟然會導致服務宕機？
- SimpleDateFormat線程不安全的5種解決方案！
- ThreadLocal不好用？那是你沒用對！
- ThreadLocal內存溢出代碼演示和原因分析！
- Semaphore自白：限流器用我就對了！
- CountDownLatch：別浪，等人齊再團！
- CyclicBarrier：人齊了，司機就可以發車了！
- synchronized 優化手段之鎖膨脹機制！
- synchronized 中的 4 個優化，你知道幾個？
- ReentrantLock 中的 4 個坑！
- 圖解：為什么非公平鎖的性能更高？
- 死鎖的 4 種排查工具！
- 死鎖終結者：順序鎖和輪詢鎖！
關注公號「Java中文社群」查看更多有意思、漲知識的 Java 並發文章。