C# 中多執行緒間的同步

一、引入

先給出一個Num類的定義

internal classNum
{
publicstaticint odd = 50000;
publicstaticint even = 10000;
}


假設現在要求輸出小於 odd 的所有奇數，輸出小於 even 的所有偶數，不考慮多執行緒時可以寫出如下的程式碼：（為了演示多執行緒時執行緒間的爭用，先把值賦給了 num，實際上這個賦值操作毫無意義 ）

//同步程式碼段
publiclongSum()
 {
Stopwatch sw = new Stopwatch();
sw.Start();
int num = 0;
for (int i = 0; i <= Num.odd; i++)
{
num = i; 
if ((i & 1) == 1)
{
Console.WriteLine($"奇數：{num}");
}
}
for (int i = 0; i <= Num.even; i++)
{
num = i;
if ((i & 1) == 0)
{
Thread.Sleep(10);
Console.WriteLine($"偶數：{num}");
}
}
sw.Stop();
return sw.ElapsedMilliseconds;
}


現在，因為耗時太長，引入多執行緒進行處理，修改為如下形式：

//NoLock Task 
privatereadonlyobject sync = new();
int num = 0; 
publicintSum()
{
Stopwatch sw = new Stopwatch();
sw.Start();
var ta =Task.Run(() =>
{
for (int i = 0; i <= Num.odd; i++)
{
num = i; //判斷條件之前賦值是為了提高觸發機率
if((i & 1) == 1)
{
Console.WriteLine($"奇數：{num}");
}
}
});
var tb = Task.Run(() =>
{
for (int i = 0; i <= Num.even; i++)
{
num = i;
if ((i & 1) == 0)
{
Thread.Sleep(10); //在此處添加延時，在 tb 執行緒等待時，num.sum的值可能已經被 ta 修改為了其他值
Console.WriteLine($"偶數：{num}");
}
}
});
Task.WaitAll(ta, tb); //為了保證任務完成，獲取執行時間
sw.Stop();
return sw.ElapsedMilliseconds;
}
}


上面這段程式碼中，我們期望執行緒ta會輸出odd以內的奇數值，執行緒tb會輸出even以內的偶數。但實際執行時會出現下圖所示的情況。

如上，當程式涉及到多執行緒的時候，在 [各執行緒間共享的數據] 總會因為執行緒間的爭用導致意料之外的情況，為此，各種語言也會提供協助執行緒間同步的特性，這裏簡單記錄一下我對c#中機制的理解。

二 、Lock

借用Lock的方法對範例進行修改，可以有兩種方式，此處展示ta部份，tb 與ta做相同修改:

//方式一：在Task內全域Lock
var ta = Task.Run(() =>
{
lock (sync)
{ 
for (int i = 0; i <= 10000; i++)
{
num = i; //判斷條件之前賦值是為了提高觸發機率
if ((i & 1) == 1)
{
Console.WriteLine($"奇數：{num}");
}
}
}
});
//方式二: 為每一次For迴圈Lock
var ta = Task.Run(() =>
{
//lock (sync)
for (int i = 0; i <= Num.odd; i++)
{
lock(sync)
{
num = i; //判斷條件之前賦值是為了提高觸發機率
if ((i & 1) == 1)
{
Console.WriteLine($"奇數：{num}");
}
}
}
});

上述的兩種方式中，

說回Lock本身，網上有很多文章介紹， lock 只是一個語法糖，編譯器會將其轉換為對 monitor 的呼叫。

IL程式碼如下圖：

可以看到，編譯器會幫我們構建try塊，並在finally塊呼叫Monitor.Exit方法。若要獲取更精細的控制，可以自己呼叫Monitor進行使用。

三、Monitor

monitor 與 lock 相比更為靈活，可以使用**IsEntered(object)**判斷當前執行緒是否獲取到了sync的釘選，可以使用 **TryEnter()**嘗試獲取排它鎖，也可以呼叫多載方法指定等待時間。

需要指出的是，

1、所有等待獲取釘選的執行緒會處於阻塞狀態。

2、在等待獲取釘選的執行緒上執行Thread.Interrupt會中斷當前執行緒的等待並丟擲ThreadInterruptedException的異常

3、monitor與lock釘選的物件sync必須為 參照型別 ，檢視反編譯的程式碼會發現在每一次lock之前，會將sync賦值給一個object物件，如果sync為值型別，則會被裝箱為一個新的物件。

4、以釘選For迴圈為例，在定義釘選物件時，可以定義為

public classLockFor
{
privatereadonlyobject sync = new();
....
}


或者定義為

public classLockForStaticSync
{
privatestaticreadonlyobject sync = new();
...
}


後者添加一個 static 修飾詞將其定義靜態唯讀物件。

我們知道，static 修飾的變量並不屬於物件，而是歸屬於 class 本身，按照我的理解來說，如果在多個執行緒中都例項化了 LockForStaticSync 的物件，並同時呼叫一個 SUM 方法時，不論執行緒間是否會發生沖突，只能有一個執行緒取得sync的釘選繼續執行，其他的執行緒會處於阻塞等待狀態。

而如果是第一種定義方式，則多個執行緒的多個物件間，不會產生沖突，所有執行緒都可以執行。

此例的場景下，多例項呼叫時，方式一應該會有速度上的絕對優勢。為此我刪除了任務內打印及等待的部份，執行了如下內容進行驗證：

privatestaticvoidInvocation()
{
Stopwatch sw = new Stopwatch();
sw.Start();
var t1 = new TaskFactory().StartNew(() => new LockFor().Sum());
var t2 = new TaskFactory().StartNew(() => new LockFor().Sum());
var t3 = new TaskFactory().StartNew(() => new LockFor().Sum());
Task.WaitAll(t1, t2, t3);
sw.Stop();
Console.WriteLine(sw.ElapsedMilliseconds);
sw.Restart();
var t4 = new TaskFactory().StartNew(() => new LockForStaticSync().Sum());
var t5 = new TaskFactory().StartNew(() => new LockForStaticSync().Sum());
var t6 = new TaskFactory().StartNew(() => new LockForStaticSync().Sum());
Task.WaitAll(t4, t5, t6);
sw.Stop();
Console.WriteLine(sw.ElapsedMilliseconds);
}

但實際執行結果是出乎意料的，多次執行後，類變量鎖的執行速度遠超物件變量的方式，這是為什麽呢？

思考之後，考慮到類變量省略了每次建立釘選物件的時間，而數量較少的迴圈次數可能無法彌補這個時間差，於是我逐漸調大Num類中定義的變量。隨著迴圈次數的增加，也就是方法執行時間的增加，物件變量的優勢逐漸顯現

既然物件變量有速度上的優勢，而使用過程中又不可避免的會出現多方呼叫的情況，那麽是不是應該一直選擇定義為物件變量呢？其實不然，比如靜態類中需要的釘選物件，全域緩存字典，檔操作幫助類都應該是靜態鎖，更多場景，歡迎補充。

四、Interlocked

Interlocked類中的方法可以實作原子操作，透過作業系統及硬體CPU級別的控制，確保CPU在執行當前操作時不會被中斷，這個類裏面提供了一些簡單的方法，如Add,Increment等。

五、Semaphore

號誌是一種計數的互斥釘選。什麽意思呢？以Monitor來講，從monitor.enetr開始，到monitor.exit為止，被包裹著的這一段程式碼，同一時刻只能由一個執行緒存取，而 Semaphore 可以定義同時存取某些資源的執行緒數量，即允許多執行緒同時存取被保護的程式碼。

Semaphore有三種簽名的建構函式，其中 Semaphore(int initialCount, int maximumCount) 的參數指定最初釋放的號誌可用數量與最大量，兩者的差值歸建立號誌的執行緒所有。

以下內容來自 MSDN

classSemaphoreTest
{
// A semaphore that simulates a limited resource pool.
privatestatic Semaphore _pool;
// 協助設定執行緒休眠時間.
privatestaticint _padding;
publicstaticvoidMain()
{
// 這裏是建立了最大可以有三個存取執行緒的號誌，但此時可用為0，需要當前執行緒釋放以後才可用。
// 如果此處設定可用量非0，主執行緒釋放時仍然傳遞了3，程式執行過程中會出現SemaphoreFullException的異常
_pool = new Semaphore(0, 3);
for (int i = 1; i <= 5; i++)
{
Thread t = new Thread(new ParameterizedThreadStart(Worker));
t.Start(i);
}
// 主執行緒休眠，讓其他執行緒執行到等待號誌的狀態
Thread.Sleep(500);
//主執行緒呼叫Release(3)，將可用量設定為最大值，正在等待的執行緒會獲得訊號
Console.WriteLine("Main thread calls Release(3)."); 
_pool.Release(3);
Console.WriteLine("Main thread exits.");
Console.ReadLine();
}
privatestaticvoidWorker(object num)
{
// 阻塞當前執行緒，直到獲取到號誌
Console.WriteLine("Thread {0} begins " +"and waits for the semaphore.", num);
_pool.WaitOne(); 
// 每一個執行緒等待的時間間隔參數
int padding = Interlocked.Add(ref _padding, 100);
Console.WriteLine("Thread {0} enters the semaphore.", num);
//與padding共用，讓輸出更有順序
Thread.Sleep(1000 + padding);
Console.WriteLine("Thread {0} releases the semaphore.", num);
Console.WriteLine("Thread {0} previous semaphore count: {1}",num, _pool.Release()); }
}









目前為止，提到的內容均是在同一個行程內同步的方法，號誌作為一個系統級的存在，是可以幫助我們實作行程間同步的，只需要在建立 Seamphore 物件的例項時，為號誌指定名稱即可。

另外特別需要註意的是，號誌是可重入的，簡單說就是可以在一個執行緒內執行多次 WaitOne() 方法，多次呼叫時，如果處理不好，就可能出現意外情況，修改 Worker 為下方內容

privatestaticvoidWorker(object num)
{
Console.WriteLine("Thread {0} begins " +
"and waits for the semaphore.", num);
_pool.WaitOne();
Console.WriteLine($"{num} getOne");
// A padding interval to make the output more orderly.
int padding = Interlocked.Add(ref _padding, 100);
_pool.WaitOne(); //此處添加一處呼叫
Console.WriteLine("Thread {0} enters the semaphore.", num);
Thread.Sleep(1000 + padding);
Console.WriteLine("Thread {0} releases the semaphore.", num);
Console.WriteLine("Thread {0} previous semaphore count: {1}",
num, _pool.Release());
//這裏應該再呼叫一次_pool.Release()，或者在上一句傳入參數，改為 _pool.Release(2)
}


建議實際執行一下，檢視執行的表現，執行過後會發現，任務執行緒全部阻塞在了第二處 WaitOne，導致程式無法向下執行。

因此，重入必須要謹慎，而且結束時必須要釋放等量的釘選數值，如果執行了多余的Release()，最終程式在執行過程中會出現

SemaphoreFullException的異常

六、Event

與號誌一樣，事件也是一個系統範圍內的資源同步方法。

又分為ManualResetEvent，AutoResetEvent，CountdownEvent以及ManualResetEventSlim，在構建物件例項時若傳入了name參數，代表這是一個可以跨行程的系統級同步事件。

以 ManualResetEvent 為例，該類有 signaled 和 nonsignaled 兩種狀態，這兩種狀態透過例項化物件時的 布爾型別 參數決定， TRUE 就是signaled, False 相反

文件中常見的轉譯是 發出訊號的狀態 和 未發出訊號的狀態 ，微軟官網的機翻是終止狀態和非終止狀態，還有一些釋放執行緒之類的描述，直觀上難以理解。其實就是改變狀態而已，還不如英文的好理解。

ManualResetEvent 的基礎類別 EventWaitHandle 中提供了Set() 和Reset() 方法，用於改變狀態，Set 將事件修改為 signaled ，Reset重設為 nonsignaled

這裏說一下Set和Reset，這兩個方法是改變了事件的狀態，並不是一個瞬時性的動作，也就意味著在呼叫Set後，呼叫Reset之前，事件都處於 signaled 狀態（AutoResetEvent會自動呼叫Reset重設事件狀態）

WaitHandle 類中提供了眾多等待訊號的方法，EventWaitHandle 繼承自WaitHandle， ManualResetEvent中也可以呼叫WaitOne等方法。

回頭再看一下號誌 Semaphore 的範例，也呼叫 WaitOne 等待訊號，因為它也繼承自 Waithandler。

有了上面的基礎，可以看一下下面 MSDN 的範例

privatestatic ManualResetEvent mre = new ManualResetEvent(false);
staticvoidMain()
{
for (int i = 0; i <= 2; i++)
{
Thread t = new Thread(ThreadProc);
t.Name = "Thread_" + i;
t.Start();
}
Thread.Sleep(500);
Console.WriteLine(@"執行緒012都會處於等待狀態，直至 mre修改狀態，按 Enter繼續");
Console.ReadLine();
mre.Set();
Thread.Sleep(500);
Console.WriteLine(@"呼叫了mre.set後，事件處於 signaled 狀態，下一個執行緒不會被阻塞");
Console.ReadLine();
for (int i = 3; i <= 4; i++)
{
Thread t = new Thread(ThreadProc);
t.Name = "Thread_" + i;
t.Start();
}
Thread.Sleep(500);
Console.WriteLine("呼叫mre.reset後，事件處於nonsignaled狀態，此時會被阻塞");
Console.ReadLine();
mre.Reset();
Thread t5 = new Thread(ThreadProc);
t5.Name = "Thread_5";
t5.Start();
Thread.Sleep(500);
Console.WriteLine("\nPress Enter to call Set() and conclude the demo.");
Console.ReadLine();
mre.Set();
Console.ReadLine();
}
privatestaticvoidThreadProc()
{
string name = Thread.CurrentThread.Name;
Console.WriteLine(name + " starts and calls mre.WaitOne()");
mre.WaitOne();
Console.WriteLine(name + " ends.");
}









七、Barrier

Barrier 是一個有意思的類，可以使多個任務能夠采用並列方式依據某種演算法在多個階段中協同工作。

Enables multiple tasks to cooperatively work on an algorithm in parallel through multiple phases.

大白話來講，借助於這個物件，可以管控多個並列任務間的合作關系。

Barrier的建構函式中可以傳入並列任務的數量（participantCount），也可以透過 AddParticipant，RemoveParticipant 動態地調整參與者的數量。另外可以透過 CurrentPhaseNumber 得知當前是第幾個參與者，透過 ParticipantsRemaining 知道還有幾個參與者未到達任務點。

還可以傳入一個可空委托，這個委托會在接收到所有參與者執行緒發出訊號後執行。

就好像幾個人一起玩遊戲闖關，關卡boss需要所有人一起才能打敗，照顧到每個人的遊戲理解不同，規定每個人可以按照自己的安排推進遊戲進度。

在這個比喻中，每個人都是單獨的執行緒，可能有人很快就抵達了關卡，但是因為關卡的性質，他必須在這裏等待其他人都到達後，大家一起打boss，打敗boss之後存檔，之後大家再各玩各的，直到下一個關卡BOSS。

Barrier就承擔了boss的任務，他負責讓所有執行緒抵達某一個預設的點後再一起放行。放行之後，會執行初始化物件時傳入的委托，比如上方說的存檔。只是我們可以透過委托，更靈活地指定要進行的操作。

而所謂的預設點，其實就是呼叫 SignalAndWait，發出訊號並等待其他執行緒發出訊號的程式碼

如果你暫時沒有理解我上面的比喻，那就看一下下面的程式碼吧，程式碼是在MSDN拿來的，我自己加了幾個console語句，可以執行看看效果

publicstaticvoidBarriers()
{
int count = 0;
//初始化 3 個參與者，傳入委托
Barrier barrier = new Barrier(3, (b) =>
{
Console.WriteLine("Post-Phase action: count={0}, phase={1},threadid={2}", count, b.CurrentPhaseNumber,Thread.CurrentThread.ManagedThreadId);
if (b.CurrentPhaseNumber == 2)
thrownew Exception("D'oh!");
});
barrier.AddParticipants(2);
barrier.RemoveParticipant(); //剩下4個
Console.WriteLine($"主Thread:{Thread.CurrentThread.ManagedThreadId}");
// This is the logic run by all participants
Action action = () =>
{
Console.WriteLine($"action1 Thread:{Thread.CurrentThread.ManagedThreadId}");
Interlocked.Increment(ref count);
barrier.SignalAndWait(); // during the post-phase action, count should be 4 and phase should be 0
Console.WriteLine($"action2 Thread:{Thread.CurrentThread.ManagedThreadId}");
Interlocked.Increment(ref count);
barrier.SignalAndWait(); // during the post-phase action, count should be 8 and phase should be 1
Console.WriteLine($"action3 Thread:{Thread.CurrentThread.ManagedThreadId}");
// 執行緒3會引發委托裏丟擲的異常，異常資訊所有執行緒可見
Interlocked.Increment(ref count);
try
{
barrier.SignalAndWait();
}
catch (BarrierPostPhaseException bppe)
{
Console.WriteLine("Caught BarrierPostPhaseException: {0}", bppe.Message);
}
Console.WriteLine($"action4 Thread:{Thread.CurrentThread.ManagedThreadId}");
// The fourth time should be hunky-dory
Interlocked.Increment(ref count);
barrier.SignalAndWait(); // during the post-phase action, count should be 16 and phase should be 3
Console.WriteLine($"action5 Thread:{Thread.CurrentThread.ManagedThreadId}");
};
// 啟動與 Barrier設定數量相同的任務，如果啟動數目超過設定值，會引發如下異常
//"System.InvalidOperationException: The number of threads using the barrier exceeded the total number of registered participants."
Parallel.Invoke(action, action, action, action);
Console.WriteLine($"主Thread2:{Thread.CurrentThread.ManagedThreadId}");
// It's good form to Dispose() a barrier when you're done with it.
barrier.Dispose();
}










執行效果如下：

可以在24行，27行打斷點，借助上面的比喻，理解一下Barrier的用法。

最後要聲明的是Barrier的public protected 成員是執行緒安全的，可以跨執行緒使用。但是Dispose是非執行緒安全的，意味著一旦呼叫，所有執行緒都會受到影響，應該在任務程式碼之外執行，另外既然有dispose的方法，就要註意使用完畢後呼叫該方法釋放資源。

八、ReaderWriterLockSlim

讀寫鎖，允許多個執行緒處於讀取模式，允許一個執行緒處於具有獨占釘選許可權的寫入模式，並且允許具有讀取存取許可權的一個執行緒處於可升級讀取模式，在該模式下，執行緒可以升級到寫入模式，而無需放棄對資源的讀取存取許可權。

讀寫鎖具有三種模式，讀鎖，寫鎖，可升級的讀鎖

讀鎖可以透過 EnterReadLock 進入，透過 ExitReadLock 結束，寫鎖類似EnterWriteLock，ExitWriteLock可升級的讀鎖是指可以直接由讀轉換為寫模式的狀態，EnterUpgradeableReadLock及ExitUpgradeableReadLock

與其他同步物件相同，讀寫鎖需要正確的進行釋放，不然會引發問題

讀寫鎖初始化時，可以傳入 LockRecursionPolicy 指定遞迴狀態，預設的建構函式為NoRecursion，微軟官網並不建議新手使用遞迴策略，因為這具有更高的復雜性，而且容易帶來死結的問題，我自己也沒有用過遞迴策略。

與 ReaderWriterLock 相比， ReaderWriterLockSlim 是被推薦使用的物件

對於 ReaderWriterLockSlim 的使用，我在園子裏有個提問請教ReaderWriterLockSlim的問題，建議轉過去看下，我這邊就不放程式碼了，當時找到了另一篇文章解答了我的疑惑，地址也貼在這裏 C# ReaderWriterLockSlim 實作 - dz45693.

這裏面就有一些需要註意的點，這幾個點全部是摘抄自上面那篇文章，請知悉：

具體的程式碼範例請參考我的提問及 dz45693 的文章

九、Mutex

Mutex 同Event，Semaphore類似，可以跨行程同步內容，定義跨行程的Mutex只需要在初始化時為其指定名字即可；都繼承自 WaitHandle ，所以也有waitone的方法可以呼叫。

使用上與 Monitor 類似，屬於互斥鎖，所以在任務的最後必須要呼叫 ReleaseMutex() 。

Mutex 實作了IDispose介面，所以需要在finally塊內呼叫 Dispose() 方法。

Mutex可以用來限定winform程式只能有一個例項執行，例項如下：

staticvoidMain()
 {
bool runone;
//獲取名為 single_test的互斥的初始所有權，runone指定是否成功
Mutex run = new Mutex(true, "single_test", out runone);
if (runone) //true代表當前未建立改互斥
{
run.ReleaseMutex();
Application.EnableVisual styles();
Application.SetCompatibleTextRenderingDefault(false);
FrmRemote frm = new FrmRemote();
int hdc = frm.Handle.ToInt32(); // write to ...
Application.Run(frm);
IntPtr a = new IntPtr(hdc);
}
else
{
MessageBox.Show("已經執行了一個例項了。");
}
 }


十、ThreadLocal ，AsyncLocal，Volatile

在定義一個類時，有時會定義全域變量，如果在編寫類時未考慮在多執行緒中使用，那麽在類中定義的全域變量很有可能會因為多執行緒呼叫引發異常，比如文章開頭的引子中，將 num 定義為了類的全域變量，造成多執行緒呼叫時出現錯誤的情況，所以需要減少全域變量的使用。但是，有時候，我們又不得不借助全域變量幫助我們實作需求，這時候就可以考慮上面提到的這幾個。

我們可能希望定義的變量對每個執行緒是唯一的，這時候就可以借助 ThreadLocal ，如果是使用了async，await的寫法，因為在await之後執行執行緒會發生變化，這時候就可以使用 AsyncLocal ，只是需要註意一下變量在父子行程間的傳遞關系是怎麽樣的。

具體可看 AsnycLocal與ThreadLocal

至於 Volatile ，在 c # 中，對 volatile 欄位使用修飾詞可保證對該欄位的每個存取都是揮發性記憶體操作。我們知道vs在release模式下編譯時會對程式碼進行最佳化，最佳化的過程中可能會修改一些內容， volatile 修飾的欄位則不會被編譯器進行最佳化。除此之外，多執行緒協作時，希望對某一個變量的修改可以立即反饋到其他執行緒中，這時候也可以借助 volatile，但 volatile 修飾詞不能套用於陣列元素。Volatile.Read和 Volatile.Write 方法可用於陣列元素。

volatile 值立馬反饋到其他執行緒是因為處理被標記欄位時，處理器不會使用緩存，而是每次都去記憶體裏讀取該欄位。

至於處理器緩存之類的，如果有興趣，可以自行了解。

AsyncLocal 和 volatile 我自己並沒有實際用過，只是在網上看了一些內容，在官方文件看了一點內容，如有需要建議自行搜尋。

十一、有意思的範例

記得之前看過一個例子，兩個執行緒迴圈輸出文字，不記得在哪看的了，試著寫一下

public classSample
 { //先執行的執行緒設定為 true
ManualResetEvent even = new ManualResetEvent(true);
ManualResetEvent odd = new ManualResetEvent(false);
publicvoidSum()
{
var ta = Task.Run(() => PrintEven(even, odd));
var tb = Task.Run(() => PrintOdd (even,odd));
}
//等待自己的訊號，控制另一個執行緒的訊號
publicvoidPrintEven(EventWaitHandle evenHandle, EventWaitHandle oddHandle)
{
string design = "偶數";
for (int i = 0; i <= 20; i++)
{
evenHandle.WaitOne();
if ((i & 1) == 0)
{
Console.WriteLine($"{design}：{i}");
evenHandle.Reset();
oddHandle.Set();
}
}
}
publicvoidPrintOdd(EventWaitHandle evenHandle, EventWaitHandle oddHandle)
{
string design = "奇數";
for (int i = 0; i <= 20; i++)
{
oddHandle.WaitOne();
if ((i & 1) == 1)
{
Console.WriteLine($"{design}：{i}");
oddHandle.Reset();
evenHandle.Set();
}
}
}
 }

流水賬似的寫下了任務同步的實作方法，於我自己而言，對文章中提到的內容加深了很多理解，希望對讀到文章的你也有幫助。

轉自：化雲隨風

連結：cnblogs.com/imzx/p/15351165.html