輪詢鎖使用時遇到的問題與解決方案!
作者 | 王磊
來源 | Java中文社群（ID：javacn666）
轉(zhuǎn)載請聯(lián)系授權(quán)（微信ID：GG_Stone
當我們遇到死鎖之后，除了可以手動重啟程序解決之外，還可以考慮是使用順序鎖和輪詢鎖，這部分的內(nèi)容可以參考我的上一篇文章，這里就不再贅述了。然而，輪詢鎖在使用的過程中，如果使用不當會帶來新的嚴重問題，所以本篇我們就來了解一下這些問題，以及相應的解決方案。
問題演示
當我們沒有使用輪詢鎖之前，可能會出現(xiàn)這樣的問題：
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(); // 創(chuàng)建鎖 A
Lock lockB = new ReentrantLock(); // 創(chuàng)建鎖 B
// 創(chuàng)建線程 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(); // 運行線程
// 創(chuàng)建線程 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(); // 運行線程
}
}
以上代碼的執(zhí)行結(jié)果如下：
從上述結(jié)果可以看出，此時程序中出現(xiàn)了線程相互等待，并嘗試獲取對方（鎖）資源的情況，這就是典型的死鎖問題了。
簡易版輪詢鎖
當出現(xiàn)死鎖問題之后，我們就可以使用輪詢鎖來解決它了，它的實現(xiàn)思路是通過輪詢的方式來獲取多個鎖，如果中途有任意一個鎖獲取失敗，則執(zhí)行回退操作，釋放當前線程擁有的所有鎖，等待下一次重新執(zhí)行，這樣就可以避免多個線程同時擁有并霸占鎖資源了，從而直接解決了死鎖的問題，簡易版的輪詢鎖實現(xiàn)如下：
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(); // 創(chuàng)建鎖 A
Lock lockB = new ReentrantLock(); // 創(chuàng)建鎖 B
// 創(chuàng)建線程 1(使用輪詢鎖)
Thread t1 = new Thread(new Runnable() {
@Override
public void run() {
// 調(diào)用輪詢鎖
pollingLock(lockA, lockB);
}
});
t1.start(); // 運行線程
// 創(chuàng)建線程 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.");
}
}
// 等待一秒再繼續(xù)執(zhí)行
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
以上代碼的執(zhí)行結(jié)果如下：
從上述結(jié)果可以看出，當我們在程序中使用輪詢鎖之后就不會出現(xiàn)死鎖的問題了，但以上輪詢鎖也并不是完美無缺的，下面我們來看看這個輪詢鎖會有什么樣的問題？
問題1：死循環(huán)
以上簡易版的輪詢鎖，如果遇到有一個線程一直霸占或者長時間霸占鎖資源的情況，就會導致這個輪詢鎖進入死循環(huán)的狀態(tài)，它會嘗試一直獲取鎖資源，這樣就會造成新的問題，帶來不必要的性能開銷，具體示例如下。
反例
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(); // 創(chuàng)建鎖 A
Lock lockB = new ReentrantLock(); // 創(chuàng)建鎖 B
// 創(chuàng)建線程 1(使用輪詢鎖)
Thread t1 = new Thread(new Runnable() {
@Override
public void run() {
// 調(diào)用輪詢鎖
pollingLock(lockA, lockB);
}
});
t1.start(); // 運行線程
// 創(chuàng)建線程 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 {
// 如果此處代碼未執(zhí)行，線程 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.");
}
}
// 等待一秒再繼續(xù)執(zhí)行
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
以上代碼的執(zhí)行結(jié)果如下：
從上述結(jié)果可以看出，線程 1 輪詢鎖進入了死循環(huán)的狀態(tài)。
優(yōu)化版
針對以上死循環(huán)的情況，我們可以改進的思路有以下兩種：
添加最大次數(shù)限制：如果經(jīng)過了 n 次嘗試獲取鎖之后，還未獲取到鎖，則認為獲取鎖失敗，執(zhí)行失敗策略之后終止輪詢（失敗策略可以是記錄日志或其他操作）； 添加最大時長限制：如果經(jīng)過了 n 秒嘗試獲取鎖之后，還未獲取到鎖，則認為獲取鎖失敗，執(zhí)行失敗策略之后終止輪詢。
以上策略任選其一就可以解決死循環(huán)的問題，出于實現(xiàn)成本的考慮，我們可以采用輪詢最大次數(shù)的方式來改進輪詢鎖，具體實現(xià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(); // 創(chuàng)建鎖 A
Lock lockB = new ReentrantLock(); // 創(chuàng)建鎖 B
// 創(chuàng)建線程 1(使用輪詢鎖)
Thread t1 = new Thread(new Runnable() {
@Override
public void run() {
// 調(diào)用輪詢鎖
pollingLock(lockA, lockB, 3);
}
});
t1.start(); // 運行線程
// 創(chuàng)建線程 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：最大輪詢次數(shù)
*/
public static void pollingLock(Lock lockA, Lock lockB, int maxCount) {
// 輪詢次數(shù)計數(shù)器
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.");
}
}
// 判斷是否已經(jīng)超過最大次數(shù)限制
if (count++ > maxCount) {
// 終止循環(huán)
System.out.println("輪詢鎖獲取失敗,記錄日志或執(zhí)行其他失敗策略");
return;
}
// 等待一秒再繼續(xù)嘗試獲取鎖
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
以上代碼的執(zhí)行結(jié)果如下：
從以上結(jié)果可以看出，當我們改進之后，輪詢鎖就不會出現(xiàn)死循環(huán)的問題了，它會嘗試一定次數(shù)之后終止執(zhí)行。
問題2：線程餓死
我們以上的輪詢鎖的輪詢等待時間是固定時間，如下代碼所示：
// 等待 1s 再嘗試獲?。ㄝ喸儯╂i
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(); // 創(chuàng)建鎖 A
Lock lockB = new ReentrantLock(); // 創(chuàng)建鎖 B
// 創(chuàng)建線程 1(使用輪詢鎖)
Thread t1 = new Thread(new Runnable() {
@Override
public void run() {
// 調(diào)用輪詢鎖
pollingLock(lockA, lockB, 3);
}
});
t1.start(); // 運行線程
// 創(chuàng)建線程 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(); // 釋放鎖
}
// 等待一秒之后繼續(xù)執(zhí)行
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
});
t2.start(); // 運行線程
}
/**
* 輪詢鎖
*/
public static void pollingLock(Lock lockA, Lock lockB, int maxCount) {
// 循環(huán)次數(shù)計數(shù)器
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.");
}
}
// 判斷是否已經(jīng)超過最大次數(shù)限制
if (count++ > maxCount) {
// 終止循環(huán)
System.out.println("輪詢鎖獲取失敗,記錄日志或執(zhí)行其他失敗策略");
return;
}
// 等待一秒再繼續(xù)嘗試獲取鎖
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
以上代碼的執(zhí)行結(jié)果如下：
從上述結(jié)果可以看出，線程 1（輪詢鎖）一直未成功獲取到鎖，造成這種結(jié)果的原因是：線程 1 每次輪詢的等待時間為固定的 1s，而線程 2 也是相同的頻率，每 1s 獲取一次鎖，這樣就會導致線程 2 會一直先成功獲取到鎖，而線程 1 則會一直處于“餓死”的情況，執(zhí)行流程如下圖所示：
優(yōu)化版
接下來，我們可以將輪詢鎖的固定等待時間，改進為固定時間 + 隨機時間的方式，這樣就可以避免因為獲取鎖的頻率一致，而造成輪詢鎖“餓死”的問題了，具體實現(xiàn)代碼如下：
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(); // 創(chuàng)建鎖 A
Lock lockB = new ReentrantLock(); // 創(chuàng)建鎖 B
// 創(chuàng)建線程 1(使用輪詢鎖)
Thread t1 = new Thread(new Runnable() {
@Override
public void run() {
// 調(diào)用輪詢鎖
pollingLock(lockA, lockB, 3);
}
});
t1.start(); // 運行線程
// 創(chuàng)建線程 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(); // 釋放鎖
}
// 等待一秒之后繼續(xù)執(zhí)行
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
});
t2.start(); // 運行線程
}
/**
* 輪詢鎖
*/
public static void pollingLock(Lock lockA, Lock lockB, int maxCount) {
// 循環(huán)次數(shù)計數(shù)器
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.");
}
}
// 判斷是否已經(jīng)超過最大次數(shù)限制
if (count++ > maxCount) {
// 終止循環(huán)
System.out.println("輪詢鎖獲取失敗,記錄日志或執(zhí)行其他失敗策略");
return;
}
// 等待一定時間(固定時間 + 隨機時間)之后再繼續(xù)嘗試獲取鎖
try {
Thread.sleep(300 + rdm.nextInt(8) * 100); // 固定時間 + 隨機時間
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
以上代碼的執(zhí)行結(jié)果如下：
從上述結(jié)果可以看出，線程 1（輪詢鎖）加入隨機等待時間之后就不會出現(xiàn)線程餓死的問題了。
總結(jié)
本文我們介紹了輪詢鎖的用途，用于解決死鎖問題，但簡易版的輪詢鎖在某些情況下會造成死循環(huán)和線程餓死的問題，因此我們對輪詢鎖進行了優(yōu)化，給輪詢鎖加入了最大輪詢次數(shù)，以及隨機輪詢等待時間，這樣就可以解決因為引入輪詢鎖而造成的新問題了，這樣就可以愉快的使用它來解決死鎖的問題了。
參考 & 鳴謝
《Java并發(fā)編程實戰(zhàn)》
---END---
原創(chuàng)并發(fā)文章推薦
評論
圖片
表情