綜述

在遇到線程安全問題的時候，我們會使用加鎖機制來確保線程安全，但如果過度地使用加鎖，則可能導(dǎo)致鎖順序死鎖(Lock-Ordering Deadlock)。或者有的場景我們使用線程池和信號量來限制資源的使用，但這些被限制的行為可能會導(dǎo)致資源死鎖(Resource DeadLock)。
我們知道Java應(yīng)用程序不像數(shù)據(jù)庫服務(wù)器，能夠檢測一組事務(wù)中死鎖的發(fā)生，進而選擇一個事務(wù)去執(zhí)行；在Java程序中如果遇到死鎖將會是一個非常嚴重的問題，它輕則導(dǎo)致程序響應(yīng)時間變長，系統(tǒng)吞吐量變小；重則導(dǎo)致應(yīng)用中的某一個功能直接失去響應(yīng)能力無法提供服務(wù)，這些后果都是不堪設(shè)想的。因此我們應(yīng)該及時發(fā)現(xiàn)和規(guī)避這些問題。
?

死鎖產(chǎn)生的條件

死鎖的產(chǎn)生有四個必要的條件

• 互斥使用，即當(dāng)資源被一個線程占用時，別的線程不能使用
• 不可搶占，資源請求者不能強制從資源占有者手中搶奪資源，資源只能由占有者主動釋放
• 請求和保持，當(dāng)資源請求者在請求其他資源的同時保持對原有資源的占有
• 循環(huán)等待，多個線程存在環(huán)路的鎖依賴關(guān)系而永遠等待下去，例如T1占有T2的資源，T2占有T3的資源，T3占有T1的資源，這種情況可能會形成一個等待環(huán)路

對于死鎖產(chǎn)生的四個條件只要能破壞其中一條即可讓死鎖消失，但是條件一是基礎(chǔ)，不能被破壞。
?

各種死鎖

鎖順序死鎖

死鎖 當(dāng)多個線程同時需要同一個鎖，但是以不同的方式獲取它們。

例如，如果線程1持有鎖A，然后請求鎖B，線程2已經(jīng)持有鎖B，然后請求鎖A，這樣一個死鎖就發(fā)生了。線程1永遠也得不到鎖B，線程2永遠也得不到鎖A。它們永遠也不知道這種情況。

public class TreeNode {TreeNode parent ? = null; ?List ? ? children = new ArrayList(); ?public synchronized void addChild(TreeNode child){if(!this.children.contains(child)) {this.children.add(child);child.setParentOnly(this);}}public synchronized void addChildOnly(TreeNode child){if(!this.children.contains(child){this.children.add(child);}}public synchronized void setParent(TreeNode parent){this.parent = parent;parent.addChildOnly(this);} ?public synchronized void setParentOnly(TreeNode parent){this.parent = parent;} }

如果一個線程（1）調(diào)用parent.addChild(child)的同時其他線程（2）在同一個parent和child實例上調(diào)用child.setParent(parent)方法，就會發(fā)生死鎖。 下面是說明這個問題的一些偽代碼：

Thread 1: parent.addChild(child); //locks parent--> child.setParentOnly(parent); ? Thread 2: child.setParent(parent); //locks child--> parent.addChildOnly()

首先，線程1調(diào)用parent.addChild(child)，因為addChild()是同步的，所以線程1會鎖住parent對象，防止其他線程獲得。

然后，線程2調(diào)用child.setParent(parent)，因為setParent()是同步的，所有線程2會鎖住child對象，防止其他線程獲得。

現(xiàn)在，parent和child對象被這兩個不同的線程鎖住。接下來，線程1嘗試調(diào)用child.setParentOnly()方法，但是child對象被線程2鎖住了，因此這個調(diào)用就會阻塞在那。線程2也嘗試調(diào)用parent.addChildOnly()方法，但是parent對象被線程1鎖住了。線程2也會阻塞在這個方法的調(diào)用上。現(xiàn)在兩個線程都在等待獲取被其他線程持有的鎖。

線程確實需要同時獲得鎖。例如，如果線程1早線程2一點點，獲得了鎖A和B，然后，線程2就會在嘗試獲取鎖B時，阻塞在那。這樣就不會有死鎖發(fā)生。由于，線程調(diào)度是不確定的，所以，我們無法準(zhǔn)確預(yù)測什么時候會發(fā)生死鎖。

@Slf4j
public class DeadLock implements Runnable {
? ? public int flag = 1;
? ? //靜態(tài)對象是類的所有對象共享的
? ? private static Object o1 = new Object(), o2 = new Object();

? ? @Override
? ? public void run() {
? ? ? ? log.info("flag:{}", flag);
? ? ? ? if (flag == 1) {
? ? ? ? ? ? synchronized (o1) {
? ? ? ? ? ? ? ? try {
? ? ? ? ? ? ? ? ? ? Thread.sleep(500);
? ? ? ? ? ? ? ? } catch (Exception e) {
? ? ? ? ? ? ? ? ? ? e.printStackTrace();
? ? ? ? ? ? ? ? }
? ? ? ? ? ? ? ? synchronized (o2) {
? ? ? ? ? ? ? ? ? ? log.info("1");
? ? ? ? ? ? ? ? }
? ? ? ? ? ? }
? ? ? ? }
? ? ? ? if (flag == 0) {
? ? ? ? ? ? synchronized (o2) {
? ? ? ? ? ? ? ? try {
? ? ? ? ? ? ? ? ? ? Thread.sleep(500);
? ? ? ? ? ? ? ? } catch (Exception e) {
? ? ? ? ? ? ? ? ? ? e.printStackTrace();
? ? ? ? ? ? ? ? }
? ? ? ? ? ? ? ? synchronized (o1) {
? ? ? ? ? ? ? ? ? ? log.info("0");
? ? ? ? ? ? ? ? }
? ? ? ? ? ? }
? ? ? ? }
? ? }

? ? public static void main(String[] args) {
? ? ? ? DeadLock td1 = new DeadLock();
? ? ? ? DeadLock td2 = new DeadLock();
? ? ? ? td1.flag = 1;
? ? ? ? td2.flag = 0;
? ? ? ? //td1,td2都處于可執(zhí)行狀態(tài)，但JVM線程調(diào)度先執(zhí)行哪個線程是不確定的。
? ? ? ? //td2的run()可能在td1的run()之前運行
? ? ? ? new Thread(td1).start();
? ? ? ? new Thread(td2).start();
? ? }
}
?

更復(fù)雜的死鎖

Thread 1 locks A, waits for B Thread 2 locks B, waits for C Thread 3 locks C, waits for D Thread 4 locks D, waits for A

線程1等待線程2，線程2等待線程3，線程3等待線程4，線程4等待線程1.

常見的數(shù)據(jù)庫死鎖

一個更復(fù)雜的死鎖發(fā)生場景，就是數(shù)據(jù)庫事務(wù)。一個數(shù)據(jù)庫可能包含許多SQL更新請求。在一個事務(wù)中，要更新一條記錄，但這條記錄被來自其它事務(wù)的更新請求鎖住了，知道第一個事務(wù)完成。在數(shù)據(jù)庫中，同一個事務(wù)內(nèi)的每條更新請求可能都會鎖住一些記錄。

如果多個事務(wù)同時運行，并且更新相同的記錄。這就會有發(fā)生死鎖的風(fēng)險。

例如：

Transaction 1, request 1, locks record 1 for update Transaction 2, request 1, locks record 2 for update Transaction 1, request 2, tries to lock record 2 for update. Transaction 2, request 2, tries to lock record 1 for update.

一個事務(wù)事先并不知道所有的它將要鎖住的記錄，所有在數(shù)據(jù)庫中檢測和預(yù)防死鎖變得更加困難。

重入鎖死

重入鎖死是一種類似于死鎖和嵌套管程失敗的情景。

如果一個線程重入獲得了一個非重入的鎖，讀寫鎖或者一些其他的同步器就會發(fā)生重入鎖死。重入意味著一個線程已經(jīng)持有了一個鎖可以再次持有它。Java的同步塊是可以沖入的。因此，下面這段代碼執(zhí)行將不會出現(xiàn)問題。

public class Reentrant{public synchronized outer(){inner();} ?public synchronized inner(){//do something} }

outer和inner方法都被聲明為synchronized，這等同于一個synchronized(this)塊。如果一個線程在outer()方法里面調(diào)用inner()方法將不會出現(xiàn)問題，因為這兩個方法都被同步在同一個管程對象"this"上。如果一個線程已經(jīng)持有了一個管程對象上的鎖，它就可以訪問同一個管程對象上所有的同步塊。這被稱作可重入。

下面Lock的實現(xiàn)是不可重入的：

public class Lock{ private boolean isLocked = false; ?public synchronized void lock()throws InterruptedException{while(isLocked){wait();}isLocked = true;} ?public synchronized void unlock(){isLocked = false;notify();} }

如果一個線程兩次調(diào)用lock()方法而在兩次調(diào)用之間沒有調(diào)用unlock()，第二次調(diào)用lock()將會阻塞。一個重入鎖死就發(fā)生了。

要避免重入鎖死你有兩種選擇：

• 編寫代碼避免獲取已經(jīng)持有的鎖

• 使用可重入鎖

使用哪種方法更適合于你的程序取決于具體的情景。可重入鎖的性能常常不如非重入鎖，而且更難實現(xiàn)，可重入鎖通常沒有不可重入鎖那么好的表現(xiàn)，而且實現(xiàn)起來復(fù)雜，但這些情況在你的項目中也許算不上什么問題。無論你的項目用鎖來實現(xiàn)方便還是不用鎖方便，可重入特性都需要根據(jù)具體問題具體分析。

如何預(yù)防死鎖 呢？

鎖排序 當(dāng)多個線程獲取同一個鎖但是以不同的順序，就會發(fā)生死鎖。

如果你確保所有的鎖一直以相同的順序被其他線程獲取，死鎖就不會發(fā)生。看下面這例子：

Thread 1: ?lock A lock B

Thread 2: ?wait for Alock C (when A locked)

Thread 3: ?wait for Await for Bwait for C

如果一個線程，像線程3，需要幾個鎖，就必須規(guī)定其獲得鎖的順序。在它獲得序列中靠前的鎖之前不能夠獲得靠后的鎖。

比如，線程2或者線程3首先要獲得鎖A，才能夠獲得鎖C。因為，線程A持有鎖A，線程2或者線程3首先必須等待直到鎖A被釋放。然后，在它們能夠獲得鎖B或者C之前，必須成功獲得鎖A。

鎖排序是一個簡單但很有效的預(yù)防死鎖的機制。但是，它僅適用于你事先知道所有的鎖的情況下。它并不適用于所有的場景。

鎖超時

另一個預(yù)防死鎖的機制是在請求鎖時設(shè)置超時時長，也就說一個線程在設(shè)置的超時時長內(nèi)如果沒有獲得鎖就會放棄。如果一個線程在給定時長內(nèi)沒有成功獲取所有必要的鎖，它將會回退，釋放所有的鎖請求，隨機等待一段時間，然后重試。隨機等待的過程中給了其他線程獲取這個鎖的一個機會，因此，這也可以讓程序在沒有鎖的情況下繼續(xù)運行。

Thread 1 locks A Thread 2 locks B ? Thread 1 attempts to lock B but is blocked Thread 2 attempts to lock A but is blocked ? Thread 1's lock attempt on B times out Thread 1 backs up and releases A as well Thread 1 waits randomly (e.g. 257 millis) before retrying. ? Thread 2's lock attempt on A times out Thread 2 backs up and releases B as well Thread 2 waits randomly (e.g. 43 millis) before retrying.

在上面的例子中，線程2將會在線程之前大約200毫秒重試去獲得鎖，所以，大體上將會獲得所有的鎖。已經(jīng)在等待的線程A一直在嘗試獲取鎖A。當(dāng)線程2完成時，線程1也將會獲得所有的鎖。

我們需要記住一個問題，上面提到的僅僅是因為一個鎖超時了，而不是說線程發(fā)生；了死鎖，這也僅僅是說這個線程獲取這個鎖花費了多少時間去完成任務(wù)。

另外，如果線程足夠多，盡管設(shè)置了超時和重試，也是會有發(fā)生死鎖的風(fēng)險。2個線程各自在重試前等待0~500毫秒也許不會發(fā)生死鎖，但如果10或者20個線程情況就不同了。這種情況發(fā)生死鎖的概率要比兩個線程的情況要高得多。

鎖超時機制存在的一個問題是在Java中在進入一個同步代碼塊時設(shè)置時長是不可能的。你不得不創(chuàng)建一個自定義的鎖相關(guān)的類或者使用在Java5中java.util.concurrency包中的并發(fā)結(jié)構(gòu)之一。

死鎖檢測

死鎖檢測是一個重量級的死鎖預(yù)防機制，主要用于在鎖排序和鎖超時都不可用的場景中。

當(dāng)一個線程請求一個鎖當(dāng)時請求被禁止時，這個線程可以遍歷鎖圖（lock graph）檢查是否發(fā)生了死鎖。例如，如果一個線程A請求鎖7，但是鎖7被線程B持有，然后，線程A可以檢測線程B是否有請求任何線程A持有的鎖。如果有，就會發(fā)生一個死鎖。

當(dāng)然，一個死鎖場景可能比兩個對象分別持有對方的鎖要復(fù)雜的鎖。線程A可能等待線程B，線程B等待線程C，線程C等待線程D，線程D等待線程A。為了檢測死鎖，線程A必須一次測試所有的被線程B請求的鎖。從線程B的鎖請求線程A到達線程C，然后又到達線程D，從上面的檢測中，線程A找到線程A自身持有的一個鎖。這樣，線程A就會知道發(fā)生了死鎖。

下面是一個被四個線程持有和請求鎖的圖。類似于這樣的一個數(shù)據(jù)結(jié)構(gòu)可以用來檢測死鎖。

那么，如果檢測到一個死鎖，這些線程可以做些什么？

一個可能的做法就是釋放所有的鎖，回退，隨機等待一段時間然后重試。這種做法與鎖超時機制非常相似除了只有發(fā)生死鎖時線程才會回退（backup）。而不僅僅是因為鎖請求超時。然而，如果大量的線程去請求同一個鎖，可能重復(fù)的發(fā)生死鎖，盡管存在回退和等待機制。

一個更好的做法就是為這些線程設(shè)置優(yōu)先級，這樣一來，就會只有一個或者一些線程在遇到死鎖時發(fā)生回退。剩下的線程繼續(xù)請求鎖假如沒有死鎖再發(fā)生。如果賦予線程的優(yōu)先級是固定的，同樣的線程總是擁有更高的優(yōu)先級。為了避免這種情況，我們可以在發(fā)生死鎖時，隨機的為線程設(shè)置優(yōu)先級。

總結(jié)

以上是生活随笔為你收集整理的多线程之死锁介绍及预防的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

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