作者 | 王磊

來源 | Java中文社群（ID：javacn666）

轉(zhuǎn)載請聯(lián)系授權(quán)（微信ID：GG_Stone

當我們遇到死鎖之后，除了可以手動重啟程序解決之外，還可以考慮是使用順序鎖和輪詢鎖，這部分的內(nèi)容可以參考我的上一篇文章，這里就不再贅述了。然而，輪詢鎖在使用的過程中，如果使用不當會帶來新的嚴重問題，所以本篇我們就來了解一下這些問題，以及相應(yīng)的解決方案。

問題演示

當我們沒有使用輪詢鎖之前，可能會出現(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)建鎖 ALock lockB = new ReentrantLock(); // 創(chuàng)建鎖 B// 創(chuàng)建線程 1Thread t1 = new Thread(new Runnable() {@Overridepublic 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)建線程 2Thread t2 = new Thread(new Runnable() {@Overridepublic 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é)果如下：

img

從上述結(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)建鎖 ALock lockB = new ReentrantLock(); // 創(chuàng)建鎖 B// 創(chuàng)建線程 1(使用輪詢鎖)Thread t1 = new Thread(new Runnable() {@Overridepublic void run() {// 調(diào)用輪詢鎖pollingLock(lockA, lockB);}});t1.start(); // 運行線程// 創(chuàng)建線程 2Thread t2 = new Thread(new Runnable() {@Overridepublic 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é)果如下：

img

從上述結(jié)果可以看出，當我們在程序中使用輪詢鎖之后就不會出現(xiàn)死鎖的問題了，但以上輪詢鎖也并不是完美無缺的，下面我們來看看這個輪詢鎖會有什么樣的問題？

問題1：死循環(huán)

以上簡易版的輪詢鎖，如果遇到有一個線程一直霸占或者長時間霸占鎖資源的情況，就會導(dǎo)致這個輪詢鎖進入死循環(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)建鎖 ALock lockB = new ReentrantLock(); // 創(chuàng)建鎖 B// 創(chuàng)建線程 1(使用輪詢鎖)Thread t1 = new Thread(new Runnable() {@Overridepublic void run() {// 調(diào)用輪詢鎖pollingLock(lockA, lockB);}});t1.start(); // 運行線程// 創(chuàng)建線程 2Thread t2 = new Thread(new Runnable() {@Overridepublic 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é)果如下：

img

從上述結(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)建鎖 ALock lockB = new ReentrantLock(); // 創(chuàng)建鎖 B// 創(chuàng)建線程 1(使用輪詢鎖)Thread t1 = new Thread(new Runnable() {@Overridepublic void run() {// 調(diào)用輪詢鎖pollingLock(lockA, lockB, 3);}});t1.start(); // 運行線程// 創(chuàng)建線程 2Thread t2 = new Thread(new Runnable() {@Overridepublic 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é)果如下：

img

從以上結(jié)果可以看出，當我們改進之后，輪詢鎖就不會出現(xiàn)死循環(huán)的問題了，它會嘗試一定次數(shù)之后終止執(zhí)行。

問題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(); // 創(chuàng)建鎖 ALock lockB = new ReentrantLock(); // 創(chuàng)建鎖 B// 創(chuàng)建線程 1(使用輪詢鎖)Thread t1 = new Thread(new Runnable() {@Overridepublic void run() {// 調(diào)用輪詢鎖pollingLock(lockA, lockB, 3);}});t1.start(); // 運行線程// 創(chuàng)建線程 2Thread t2 = new Thread(new Runnable() {@Overridepublic 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é)果如下：

img

從上述結(jié)果可以看出，線程 1（輪詢鎖）一直未成功獲取到鎖，造成這種結(jié)果的原因是：線程 1 每次輪詢的等待時間為固定的 1s，而線程 2 也是相同的頻率，每 1s 獲取一次鎖，這樣就會導(dǎo)致線程 2 會一直先成功獲取到鎖，而線程 1 則會一直處于“餓死”的情況，執(zhí)行流程如下圖所示：

img

優(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)建鎖 ALock lockB = new ReentrantLock(); // 創(chuàng)建鎖 B// 創(chuàng)建線程 1(使用輪詢鎖)Thread t1 = new Thread(new Runnable() {@Overridepublic void run() {// 調(diào)用輪詢鎖pollingLock(lockA, lockB, 3);}});t1.start(); // 運行線程// 創(chuàng)建線程 2Thread t2 = new Thread(new Runnable() {@Overridepublic 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é)果如下：

img

從上述結(jié)果可以看出，線程 1（輪詢鎖）加入隨機等待時間之后就不會出現(xiàn)線程餓死的問題了。

總結(jié)

本文我們介紹了輪詢鎖的用途，用于解決死鎖問題，但簡易版的輪詢鎖在某些情況下會造成死循環(huán)和線程餓死的問題，因此我們對輪詢鎖進行了優(yōu)化，給輪詢鎖加入了最大輪詢次數(shù)，以及隨機輪詢等待時間，這樣就可以解決因為引入輪詢鎖而造成的新問題了，這樣就可以愉快的使用它來解決死鎖的問題了。

參考 & 鳴謝

《Java并發(fā)編程實戰(zhàn)》

---END---

---

原創(chuàng)并發(fā)文章推薦

1.線程的故事：我的3位母親成就了優(yōu)秀的我！

2.線程池的7種創(chuàng)建方式，強烈推薦你用它...

3.輕量級鎖一定比重量級鎖快嗎？

4.這樣終止線程，竟然會導(dǎo)致服務(wù)宕機？

5.漫畫：如何證明sleep不釋放鎖，而wait釋放鎖？

6.池化技術(shù)到達有多牛？看了這個對比嚇我一跳！

7.求求你，別再用wait和notify了！

8.Semaphore自白：限流器用我就對了！

9.CountDownLatch：別浪，等人齊再團！

10.CyclicBarrier：人齊了，老司機就發(fā)車了！

11.Java中用戶線程和守護線程區(qū)別這么大？

12.ThreadLocal不好用？那是你沒用對！

13.ThreadLocal內(nèi)存溢出代碼演示和原因分析！

14.SimpleDateFormat線程不安全的5種解決方案！

15.synchronized 加鎖 this 和 class 的區(qū)別！

16.synchronized 優(yōu)化手段之鎖膨脹機制！

17.synchronized 中的 4 個優(yōu)化，你知道幾個？

18.ReentrantLock 中的 4 個坑！

19.圖解：為什么非公平鎖的性能更高？

20.死鎖的 4 種排查工具 ！

21.死鎖終結(jié)者：順序鎖和輪詢鎖！

---

總結(jié)

以上是生活随笔為你收集整理的轮询锁使用时遇到的问题与解决方案!的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。