[C#.NET 拾遺補漏]12：死鎖和活鎖的發(fā)生及避免
多線程編程時，如果涉及同時讀寫共享數(shù)據(jù)，就要格外小心。如果共享數(shù)據(jù)是獨占資源，則要對共享數(shù)據(jù)的讀寫進(jìn)行排它訪問，最簡單的方式就是加鎖。鎖也不能隨便用，否則可能會造成死鎖和活鎖。本文將通過示例詳細(xì)講解死鎖和活鎖是如何發(fā)生的，以及如何避免它們。
避免多線程同時讀寫共享數(shù)據(jù)
在實際開發(fā)中，難免會遇到多線程讀寫共享數(shù)據(jù)的需求。比如在某個業(yè)務(wù)處理時，先獲取共享數(shù)據(jù)（比如是一個計數(shù)），再利用共享數(shù)據(jù)進(jìn)行某些計算和業(yè)務(wù)處理，最后把共享數(shù)據(jù)修改為一個新的值。由于是多個線程同時操作，某個線程取得共享數(shù)據(jù)后，緊接著共享數(shù)據(jù)可能又被其它線程修改了，那么這個線程取得的數(shù)據(jù)就是錯誤的舊數(shù)據(jù)。我們來看一個具體代碼示例：
static?int count { get; set; }
static?void?Main(string[] args)
{
for (int i = 1; i <= 2; i++)
{
var thread = new Thread(ThreadMethod);
thread.Start(i);
Thread.Sleep(500);
}
}
static?void?ThreadMethod(object threadNo)
{
while (true)
{
var temp = count;
Console.WriteLine("線程 " + threadNo + " 讀取計數(shù)");
Thread.Sleep(1000); // 模擬耗時工作
count = temp + 1;
Console.WriteLine("線程 " + threadNo + " 已將計數(shù)增加至: " + count);
Thread.Sleep(1000);
}
}
示例中開啟了兩個獨立的線程開始工作并計數(shù)，假使當(dāng)?ThreadMethod?被執(zhí)行第 4 次的時候（即此刻?count?值應(yīng)為 4），count?值的變化過程應(yīng)該是：1、2、3、4，而實際運行時計數(shù)的的變化卻是：1、1、2、2...。也就是說，除了第一次，后面每次，兩個線程讀取到的計數(shù)都是舊的錯誤數(shù)據(jù)，這個錯誤數(shù)據(jù)我們把它叫作臟數(shù)據(jù)。
因此，對共享數(shù)據(jù)進(jìn)行讀寫時，應(yīng)視其為獨占資源，進(jìn)行排它訪問，避免同時讀寫。在一個線程對其進(jìn)行讀寫時，其它線程必須等待。避免同時讀寫共享數(shù)據(jù)最簡單的方法就是加鎖。
修改一下示例，對?count?加鎖：
static?int count { get; set; }
static?readonly?object key = new?object();
static?void?Main(string[] args)
{
...
}
static?void?ThreadMethod(object threadNumber)
{
while (true)
{
lock(key)
{
var temp = count;
...
count = temp + 1;
...
}
Thread.Sleep(1000);
}
}
這樣就保證了同時只能有一個線程對共享數(shù)據(jù)進(jìn)行讀寫，避免出現(xiàn)臟數(shù)據(jù)。
死鎖的發(fā)生
上面為了解決多線程同時讀寫共享數(shù)據(jù)問題，引入了鎖。但如果同一個線程需要在一個任務(wù)內(nèi)占用多個獨占資源，這又會帶來新的問題：死鎖。簡單來說，當(dāng)線程在請求獨占資源得不到滿足而等待時，又不釋放已占有資源，就會出現(xiàn)死鎖。死鎖就是多個線程同時彼此循環(huán)等待，都等著另一方釋放其占有的資源給自己用，你等我，我待你，你我永遠(yuǎn)都處在彼此等待的狀態(tài)，陷入僵局。下面用示例演示死鎖是如何發(fā)生的：
class?Program
{
static?void?Main(string[] args)
{
var workers = new Workers();
workers.StartThreads();
var output = workers.GetResult();
Console.WriteLine(output);
}
}
class?Workers
{
Thread thread1, thread2;
object resourceA = new?object();
object resourceB = new?object();
string output;
public?void?StartThreads()
{
thread1 = new Thread(Thread1DoWork);
thread2 = new Thread(Thread2DoWork);
thread1.Start();
thread2.Start();
}
public?string?GetResult()
{
thread1.Join();
thread2.Join();
return output;
}
public?void?Thread1DoWork()
{
lock (resourceA)
{
Thread.Sleep(100);
lock (resourceB)
{
output += "T1#";
}
}
}
public?void?Thread2DoWork()
{
lock (resourceB)
{
Thread.Sleep(100);
lock (resourceA)
{
output += "T2#";
}
}
}
}
示例運行后永遠(yuǎn)沒有輸出結(jié)果，發(fā)生了死鎖。線程 1 工作時鎖定了資源 A，期間需要鎖定使用資源 B；但此時資源 B 被線程 2 獨占，恰巧資線程 2 此時又在待資源 A 被釋放；而資源 A 又被線程 1 占用......，如此，雙方陷入了永遠(yuǎn)的循環(huán)等待中。
死鎖的避免
針對以上出現(xiàn)死鎖的情況，要避免死鎖，可以使用?Monitor.TryEnter(obj, timeout)?方法來檢查某個對象是否被占用。這個方法嘗試獲取指定對象的獨占權(quán)限，如果?timeout?時間內(nèi)依然不能獲得該對象的訪問權(quán)，則主動“屈服”，調(diào)用?Thread.Yield()?方法把該線程已占用的其它資源交還給 CUP，這樣其它等待該資源的線程就可以繼續(xù)執(zhí)行了。即，線程在請求獨占資源得不到滿足時，主動作出讓步，避免造成死鎖。
把上面示例代碼的?Workers?類的?Thread1DoWork?方法使用?Monitor.TryEnter?修改一下：
// ...（省略相同代碼）
public?void?Thread1DoWork()
{
bool mustDoWork = true;
while (mustDoWork)
{
lock (resourceA)
{
Thread.Sleep(100);
if (Monitor.TryEnter(resourceB, 0))
{
output += "T1#";
mustDoWork = false;
Monitor.Exit(resourceB);
}
}
if (mustDoWork) Thread.Yield();
}
}
public?void?Thread2DoWork()
{
lock (resourceB)
{
Thread.Sleep(100);
lock (resourceA)
{
output += "T2#";
}
}
}
再次運行示例，程序正常輸出?T2#T1#?并正常結(jié)束，解決了死鎖問題。
注意，這個解決方法依賴于線程 2 對其所需的獨占資源的固執(zhí)占有和線程 1 愿意“屈服”作出讓步，讓線程 2 總是優(yōu)先執(zhí)行。同時注意，線程 1 在鎖定?resourceA?后，由于爭奪不到?resourceB，作出了讓步，把已占有的?resourceA?釋放掉后，就必須等線程 2 使用完?resourceA?重新鎖定?resourceA?再重做工作。
正因為線程 2 總是優(yōu)先，所以，如果線程 2 占用?resourceA?或?resourceB?的頻率非常高（比如外面再嵌套一個類似?while(true)?的循環(huán) ），那么就可能導(dǎo)致線程 1 一直無法獲得所需要的資源，這種現(xiàn)象叫線程饑餓，是由高優(yōu)先級線程吞噬低優(yōu)先級線程 CPU 執(zhí)行時間的原因造成的。線程饑餓除了這種的原因，還有可能是線程在等待一個本身也處于永久等待完成的任務(wù)。
我們可以繼續(xù)開個腦洞，上面示例中，如果線程 2 也愿意讓步，會出現(xiàn)什么情況呢？
活鎖的發(fā)生和避免
我們把上面示例改造一下，使線程 2 也愿意讓步：
public?void?Thread1DoWork()
{
bool mustDoWork = true;
Thread.Sleep(100);
while (mustDoWork)
{
lock (resourceA)
{
Console.WriteLine("T1 重做");
Thread.Sleep(1000);
if (Monitor.TryEnter(resourceB, 0))
{
output += "T1#";
mustDoWork = false;
Monitor.Exit(resourceB);
}
}
if (mustDoWork) Thread.Yield();
}
}
public?void?Thread2DoWork()
{
bool mustDoWork = true;
Thread.Sleep(100);
while (mustDoWork)
{
lock (resourceB)
{
Console.WriteLine("T2 重做");
Thread.Sleep(1100);
if (Monitor.TryEnter(resourceA, 0))
{
output += "T2#";
mustDoWork = false;
Monitor.Exit(resourceB);
}
}
if (mustDoWork) Thread.Yield();
}
}
注意，為了使我要演示的效果更明顯，我把兩個線程的 Thread.Sleep 時間拉開了一點點。運行后的效果如下：
通過觀察運行效果，我們發(fā)現(xiàn)線程 1 和線程 2 一直在相互讓步，然后不斷重新開始。兩個線程都無法進(jìn)入?Monitor.TryEnter?代碼塊，雖然都在運行，但卻沒有真正地干活。
我們把這種線程一直處于運行狀態(tài)但其任務(wù)卻一直無法進(jìn)展的現(xiàn)象稱為活鎖?；铈i和死鎖的區(qū)別在于，處于活鎖的線程是運行狀態(tài)，而處于死鎖的線程表現(xiàn)為等待；活鎖有可能自行解開，死鎖則不能。
要避免活鎖，就要合理預(yù)估各線程對獨占資源的占用時間，并合理安排任務(wù)調(diào)用時間間隔，要格外小心?，F(xiàn)實中，這種業(yè)務(wù)場景很少見。示例中這種復(fù)雜的資源占用邏輯，很容易把人搞蒙，而且極不容易維護(hù)。推薦的做法是使用信號量機制代替鎖，這是另外一個話題，后面單獨寫文章講。
總結(jié)
我們應(yīng)該避免多線程同時讀寫共享數(shù)據(jù)，避免的方式，最簡單的就是加鎖，把共享數(shù)據(jù)作為獨占資源來進(jìn)行排它使用。
多個線程在一次任務(wù)中需要對多個獨占資源加鎖時，就可能因相互循環(huán)等待而出現(xiàn)死鎖。要避免死鎖，就至少得有一個線程作出讓步。即，在發(fā)現(xiàn)自己需要的資源得不到滿足時，就要主動釋放已占有的資源，以讓別的線程可以順利執(zhí)行完成。
大部分情況安排一個線程讓步便可避免死鎖，但在復(fù)雜業(yè)務(wù)中可能會有多個線程互相讓步的情況造成活鎖。為了避免活鎖，需要合理安排線程任務(wù)調(diào)用的時間間隔，而這會使得業(yè)務(wù)代碼變得非常復(fù)雜。更好的做法是放棄使用鎖，而換成使用信號量機制來實現(xiàn)對資源的獨占訪問。
-
精致碼農(nóng)
帶你洞悉編程與架構(gòu)
↑長按圖片識別二維碼關(guān)注，不要錯過網(wǎng)海相遇的緣分