C# 多線程學習筆記 - 2
本文主要針對 GKarch 相關文章留作筆記,僅在原文基礎上記錄了自己的理解與摘抄部分片段。
遵循原作者的 CC 3.0 協議。
如果想要了解更加詳細的文章信息內容,請訪問下列地址進行學習。原文章地址:https://blog.gkarch.com/threading/part2.html
一、同步概要
同步構造基本分為四種,簡單的阻塞方法、鎖構造、信號構造、非阻塞同步構造。
1.1 阻塞方法
- 阻塞方法一般是會暫停某些線程的執行,例如
Sleep()
與Join()
方法。當一個線程被阻塞的時候,會立即出讓(yields) CPU 時間片,不再消耗處理器時間。 - 通過檢查線程的
ThreadState
- 當某個線程被阻塞或者解除阻塞的時候,會進行上下文切換。
- 阻塞方法在滿足以下幾個條件的時候會進行解除。
- 阻塞條件滿足。
- 操作超時。
- 通過
Thread.Interrupt()
中斷。 - 通過
Thread.Abort()
中止。
1.2 阻塞與自旋
信號構造與鎖構造可以在某些條件被滿足前阻塞線程。另外一種方法就是通過自旋來等待條件被滿足。
自旋即通過一個循環不斷檢測條件,來偽造一個空忙狀態。
雖然自旋會造成大量的處理器時間浪費,但是它可以避免上下文切換帶來的額外開銷。
一個標準的自旋結構如下列代碼。
// 單純的自旋 while(!proceed); // 阻塞 + 自旋 while(!proceed) Thread.Sleep(10);
二、鎖
2.1 排它鎖
排它鎖的作用是為了保證線程安全,如下列代碼。如果
Go()
方法被兩個線程同時執行,則可能某個線程在執行完if
後,另一個線程已經將V2
置為0,原先線程就可能造成除數不能為 0 的異常。class Code1 { static int V1 = 1,V2 = 1; static void Go() { if(V2 != 0) Console.WriteLine(V1 / V2); V2 = 0; } } // 使用了排它鎖的代碼 class Code2 { static int V1 = 1,V2 = 1; static readonly object locker = new object(); static void Go() { lock(locker) { if(V2 != 0) Console.WriteLine(V1 / V2); V2 = 0; } } }
如果使用了
lock
語句快,則可以鎖定一個同步對象,其他競爭鎖的線程會被阻塞,直到鎖被釋放。如果有多個線程競爭鎖,則按照先到先得的隊列進行排隊,通過排它鎖可以強制線程對鎖保護的內容進行順序訪問。
在競爭鎖時被阻塞的線程,其狀態為
WaitSleepJoin
。不同的同步結構技術的性能開銷。
構造 用途 開銷 lock
( Monitor.Enter / Monitor.Exit )確保同一時間只有一個線程可以訪問資源或代碼 20 ns Mutex 確保同一時間只有一個線程可以訪問資源或代碼 1000 ns SemaphoreSlim 確保只有不超過指定數量的線程可以並發訪問資源或代碼 200 ns Semaphore 確保只有不超過指定數量的線程可以並發訪問資源或代碼 1000 ns ReaderWriterLockSlim 允許多個讀線程和一個寫線程共存 40 ns ReaderWriterLock
(已過時)允許多個讀線程和一個寫線程共存 100 ns
2.2 Monitor.Enter 與 Monitor.Exit
lock
語句塊實質上就是一個語法糖,其核心代碼就是結合try/finally
來調用Monitor.Enter()
與Monitor.Exit()
方法,並且如果在一個方法內直接調用Monitor.Exit()
會直接拋出異常。Monitor.Enter(locker); try { if(V2 != 0) Console.Writeline(V1 / V2); V2 = 0; } finally { Monitor.Exit(locker); }
上述情況可能發生鎖泄漏,因為在
Monitor.Enter()
與try/finally
語句塊之間如果發生了異常,會導致後續的try/finally
語句塊不被執行。造成無法獲得鎖,或者得到鎖之後,無法釋放造成鎖泄漏。解決鎖泄漏的方式是,CLR 4.0 當中,對於
lock
語句的翻譯則是通過一個bool
類型的lockTaken
進行解決。bool lockTaken = false; try { Monitor.Enter (locker, ref lockTaken); // 用戶代碼 ... } finally { if(lockTaken) { Monitor.Exit(locker); } }
Monitor
還提供了TryEnter()
方法,用於執行超時時間,如果超過時間沒有獲得到鎖,則返回false
。
2.3 什麽時候加鎖
需要訪問任意可寫的共享字段,下面代碼展示了線程安全與非線程安全的代碼。
class ThreadUnsafe { static int _x; static void Increment() { _x++; } static void Assign() { _x = 123; } } // 線程安全 class ThreadSafe { static readonly object _locker = new object(); static int _x; static void Increment() { lock(_locker) _X++; } static void Assign() { lock(_locker) _x++; } }
2.4 鎖與原子性
- 如果一組變量總是在一個鎖內進行讀且,即可被成為原子的讀寫。
- 例如
lock(locker) { if(x != 0) y /= x; }
就可以說x
與y
是被原子訪問的,因為這段代碼無法被其他線程分割或者搶占。 - 如果在
lock
鎖內拋出異常,將會影響鎖的原子性,這個時候就需要結合回滾機制來進行實現。
2.5 鎖嵌套
- 排他鎖是可以被嵌套的,並且只有當最外層的鎖被釋放的時候,對象才會被解鎖。
- 線程只會在最外層的
lock
語句處被阻塞。
2.6 死鎖
死鎖是當兩個甚至多個線程所等待的資源都被對方占用的時候,它們都無法執行,就會產生了死鎖。
一個標準的死鎖代碼如下,我們在 A 線程內部鎖定了
locker1
與locker2
,在主線程同同時也鎖定了locker2
與locker1
。這個時候由於排他鎖的特性,主線程與新開啟的線程都會等待對方的鎖被釋放,造成死鎖。object locker1 = new object(); object locker2 = new object(); new Thread(() => { lock(locker1) { Thread.Sleep(1000); lock(loekcer2); } }).Start(); lock(locker2) { Thread.Sleep(); lock(locker1); }
應該盡量較少對鎖的使用,更多的依靠其他的同步構造進行處理。
2.7 互斥體
- 互斥體使用
WaitOne()
方法進行加鎖,使用ReleaseMutex()
來解鎖。 - 關閉或者銷毀
Mutex
對象會自動釋放鎖,所以可以結合using
語句塊進行使用。 - 互斥體是機器範圍的,其性能比
lock
慢約 50 倍。
2.8 信號量
信號量具有一定容量,當容量滿了之後和就會拒絕其他線程占用,當有一個線程釋放資源之後,其他線程按先後順序進入。
class Program { static void Main(string[] args) { var sem = new Semaphore(); for (int i = 1; i <= 5; i++) new Thread(sem.Enter).Start(i); } } public class Semaphore { private readonly SemaphoreSlim _semaphoreSlim = new SemaphoreSlim(3); public void Enter(object id) { Console.WriteLine($"Id 為 {id} 的線程想調用本方法。"); _semaphoreSlim.Wait(); Console.WriteLine($"Id 為 {id} 的線程已經進入方法。"); Thread.Sleep(1000 * (int)id); Console.WriteLine($"Id 為 {id} 的線程正在離開方法。"); _semaphoreSlim.Release(); } }
容量為 1 的信號量與
Mutex
和lock
類似。信號量是線程無關的,任何線程都可以調用
Release()
方法釋放信號量。而Mutex
與lock
只有獲得鎖的線程才可以釋放。在 .NET 4.0 當中有一個輕量級的信號量
SemaphoreSlim
,但是不是跨進程的,開銷只有Semaphore
的四分之一。一般在某些需要限流或者是要執行比較密集的磁盤 I/O 操作,這個時候可以使用信號量進行並發限制,這樣可以改善程序整體的性能。
三、線程安全
- 如果某個程序或者方法是一個線程安全的,那麽它在任意的多線程場景中都不會存在不確定性。
- 線程安全是通過鎖,或者減少線程交互來實現的。
- 多線程當中的線程安全一般是在需要的時候才會進行實現,但是可以以犧牲粒度,將整段代碼甚至是整個對象封裝在一個排它鎖的內部。這種解決方案十分簡單有效,這種方案適用於使用了線程不安全的第三方庫代碼,並且僅適用於能夠快速執行的場景,否則會產生大量阻塞。
- 除了 CLR 定義的基本類型以外,很少有能夠在高並發需求下保證其實例是線程安全的,除了 Concurrent 下的並行集合以外。
- 可以最小化共享數據來減少線程交互,例如 Web 服務器不需要持久化並發請求的數據,是無狀態的,線程交互的時候僅需要考慮靜態字段等共享資源。
- 除了以上兩種方法之外,也可以使用自動鎖機制,集成
ContextBoundObject
類並且使用Synchronization
特性。但是這種方法很容易造成死鎖的情況,並且降低並發度。
3.1 線程安全與 BCL 類型
通過鎖可以將不安全的代碼轉換為線程安全的代碼,例如 BCL 提供的
List<T>
集合本身不是線程安全的,但是通過對一個集合實例的鎖定,我們就可以進行線程安全的操作。下面的代碼當中,我們直接使用List<int>
集合自身來加鎖,這裏對集合進行遍歷的操作也不是線程安全的,也需要加鎖進行處理,另一種方式就是通過讀寫鎖來實現避免長時間鎖定。class Program { static void Main(string[] args) { var bcl = new BCLThreadSafe(); for (int i = 0; i < 10; i++) { new Thread(bcl.AddItem).Start(); } } } public class BCLThreadSafe { private readonly List<int> _innerList = new List<int>(); public void AddItem() { lock (_innerList) { _innerList.Add(_innerList.Count); } var sb = new StringBuilder(); lock (_innerList) { foreach (var item in _innerList) { sb.Append(item).Append(','); } } Console.WriteLine(sb.ToString().TrimEnd(',')); } }
即便
List<T>
集合是線程安全的,如果我們需要使用以下代碼增加一個新的數據到集合當中。也會由於在執行if
之後,其他線程搶占修改了_list
集合,增加了一個相同的類目。在這個時候,對_list
集合的添加操作就是存在問題的。if(!_list.Contains(newItem)) _list.Add(newItem);
在高並發的環境下,對集合的訪問加鎖可能產生大量阻塞,所以進行類似操作的時候建議使用線程安全的隊列、棧、字典。
針對於靜態成員,BCL 的所有類型的靜態成員都實現了線程安全,所以開發人員在開發基礎類型或者框架的時候,應該保證靜態成員的線程安全。
大部分 BCL 類型的只讀訪問都是線程安全的,開發人員在設計類基礎類型或者框架的時候也應該遵循這個規則。
3.2 應用服務器其與線程安全
服務端經常需要使用到多線程處理客戶請求,也就意味著必須考慮線程安全。但一般來說服務端類都是無狀態的,或者為每個請求創建新的對象實例,很少存在有交互的點。
以緩存為例,假設對一個用戶表使用了靜態的字典實例進行緩存,那麽就存在線程安全的問題。下列代碼在讀取與更新鎖的時候,使用了排它鎖進行加鎖處理。但是會存在兩個線程同時訪問
GetUser()
方法的時候,都傳遞了未緩存過得數據的id
,這個時候就會去查詢兩次數據庫。雖然可以通過對整個GetUser()
加鎖,但是這樣設計的話,都會在QueryUser()
進行查詢的時候,整個獲得用戶信息的方法都被阻塞。static class UserCache { static Dictionary<int,User> _users = new Dictionary<int,User>(); internal static User GetUser(int id) { User u = null; lock(_users) { if(_users.TryGetValue(id,out u)) { return u; } } // 從數據庫查詢用戶數據 u = QueryUser(id); lock(_users)_users[id] = u; return u; } }
3.3 WPF 與 WinForm 程序的線程安全
富客戶端程序一般都是基於
DependencyObject
(WPF) 與Control
(Windows Forms),它們都具備線程親和性,即只有創建他們的線程才能夠訪問其成員。作用就是訪問 UI 對象並不需要加鎖,壞處則是如果要跨線程調用 UI 控件則需要一些比較繁瑣的步驟。
- WPF:其
Dispatcher
調用Invoke()
或BeginInvoke()
。 - Windows Forms:調用
Control()
對象的Invoke()
或BeginInvoke()
。
- WPF:其
Invoke()
與BeginInvoke()
都接收一個委托以便代替工作線程需要在 UI 線程執行的操作。前者是同步方法,在委托執行完成之前,都處於阻塞狀態。後者是異步方法,調用方立刻返回。// WPF DEMO public partial class MainWindow : Window { public MainWindow() { InitializeComponent(); new Thread(Work).Start(); } private void Work() { Thread.Sleep(5000); // 阻塞當前線程 5s 模擬耗時任務 UpdateMessage("new msg"); } private void UpdateMessage(string msg) { var action = () => txtMessage.Text = msg; Dispatcher.Invoke(action); } } // Windows Forms DEMO public partial class FormClass : Form { // ... 其他代碼 private void UpdateMessage(string msg) { var action = () => txtMessage.Text = msg; this.Invoke(action); } // ... 其他代碼 }
3.4 不可變對象
- 不可變對象擁有的不變性可以在多線程環境中最小化共享可寫狀態的問題。
四、事件等待句柄與信號同步
事件等待句柄的作用是用於進行信號同步。
信號同步即一個線程進行等待,直到其接受到其他線程通知的過程。
信號構造的開銷比較。
構造 用途 開銷 AutoResetEvent 使線程在接收到其他線程信號時解除阻塞一次。 1000 ns ManualResetEvent 使線程在接收到其他線程信號時解除阻塞,並不繼續
阻塞,直到其復位。1000 ns ManualResetEventSlim 使線程在接收到其他線程信號時解除阻塞,並不繼續
阻塞,直到其復位。40 ns CountdownEvent 使線程在收到預訂數量的信號時,解除阻塞。 40 ns Barrier 實現線程執行屏障。 80 ns Wait 和 Pulse 使線程阻塞,直到自定義條件被滿足。 Pulse/120 ns
4.1 AutoResetEvent
AutoResetEvent
的原理類似於驗票閘機,在閘機處調用WaitOne()
方法,線程就會被阻塞。插入票的動作就類似於調用Set()
方法打開閘機。任何能夠訪問這個AutoResetEvent
的非阻塞線程都可以調用Set()
方法來放行一個被阻塞的線程。AutoResetEvent
是基於EventWaitHandle
進行構造的,有兩種方法可以創建AutoResetEvent
對象。第一種即通過其構造方法var auto = new AutoResetEvent (false);
,第二種則是通過EventWaitHandle
傳遞事件類型,var auto = new EventWaitHandle (false, EventResetMode.AutoReset);
。這裏如果傳遞的是false
則會在創建後立即調用Set()
方法。class Program { public static readonly EventWaitHandle WaitHandle = new AutoResetEvent(false); static void Main(string[] args) { var testClass = new AutoResetEventTest(); new Thread(testClass.Waiter).Start(); // 主線程等待 1 秒再發送信號喚醒 Thread.Sleep(1000); WaitHandle.Set(); } } public class AutoResetEventTest { public void Waiter() { Console.WriteLine("線程開始等待..."); // 如果傳入了超時時間,超時則返回 false。 Program.WaitHandle.WaitOne(); Console.WriteLine("接受到了通知,進入閘機。"); } }
如果沒有線程等待的時候調用
Set()
方法,則等待句柄會保持初始狀態,直到有線程調用了WaitOne()
方法。為等待句柄調用
Reset()
方法可以關閉閘機,這個方法不會被阻塞。可以調用
Dispose()
方法來銷毀等待句柄,或者直接丟棄,等待 GC 進行回收。如果主線程需要向工作線程連續發送 3 個信號並結束線程,則可以通過雙向信號進行實現,其步驟大體如下。
- 啟動工作線程。
- 主線程通過事件等待句柄 A 等待工作線程就緒。
- 工作線程通過 A 句柄通知主線程就緒,工作線程通過事件等待句柄 B 等待主線程通知。
- 主線程更改某個共享數據,通過句柄 B 通知工作線程進行處理。
- 工作線程收到信號喚醒之後,輸出共享數據,並判斷是否應該結束線程。
- 循環往復三次之後,工作線程收到的共享數據為
null
,工作線程進行退出。
static void Main(string[] args) { var testObj = new MultiAutoResetEventTest(); new Thread(testObj.WorkThread).Start(); MultiAutoResetEventTest.WaitHandle_MainThread.WaitOne(); lock (MultiAutoResetEventTest.Locker) MultiAutoResetEventTest.Message = DateTime.Now.ToFileTimeUtc().ToString(); MultiAutoResetEventTest.WaitHandle_WorkThread.Set(); MultiAutoResetEventTest.WaitHandle_MainThread.WaitOne(); lock (MultiAutoResetEventTest.Locker) MultiAutoResetEventTest.Message = DateTime.Now.ToFileTimeUtc().ToString(); MultiAutoResetEventTest.WaitHandle_WorkThread.Set(); MultiAutoResetEventTest.WaitHandle_MainThread.WaitOne(); lock (MultiAutoResetEventTest.Locker) MultiAutoResetEventTest.Message = DateTime.Now.ToFileTimeUtc().ToString(); MultiAutoResetEventTest.WaitHandle_WorkThread.Set(); MultiAutoResetEventTest.WaitHandle_MainThread.WaitOne(); lock (MultiAutoResetEventTest.Locker) MultiAutoResetEventTest.Message = null; MultiAutoResetEventTest.WaitHandle_WorkThread.Set(); } } public class MultiAutoResetEventTest { public static readonly EventWaitHandle WaitHandle_MainThread = new AutoResetEvent(false); public static readonly EventWaitHandle WaitHandle_WorkThread = new AutoResetEvent(false); public static string Message = string.Empty; public static readonly object Locker = new object(); public void WorkThread() { while (true) { WaitHandle_MainThread.Set(); WaitHandle_WorkThread.WaitOne(); lock (Locker) { if (Message == null) return; Console.WriteLine($"收到主線程的消息,內容為: {Message}"); } } } }
生產消費者隊列的構成如下所描述的一致。
- 構建一個隊列,用於存放需要執行的工作項。
- 如果有新的任務需要執行,將其放在隊列當中。
- 一個或多個工作線程在後臺執行,從隊列中拿取工作項執行,將其消費。
生產/消費者隊列可以精確控制工作線程的數量,CLR 的線程池就是一種生產/消費者隊列。
結合
AutoResetEvent
事件等待句柄,我們可以很方便地實現一個生產/消費者隊列。class Program { static void Main(string[] args) { using (var queue = new ProducerConsumerQueue()) { queue.EnqueueTask("Hello"); for (int i = 0; i < 10; i++) { queue.EnqueueTask($"{i}"); } queue.EnqueueTask("End"); } } } public class ProducerConsumerQueue : IDisposable { private readonly EventWaitHandle _waitHandle = new AutoResetEvent(false); private readonly object _locker = new object(); private readonly Queue<string> _taskQueue = new Queue<string>(); private readonly Thread _workThread; public ProducerConsumerQueue() { _workThread = new Thread(Work); _workThread.Start(); } public void EnqueueTask(string task) { // 向隊列當中插入任務,加鎖保證線程安全 lock (_locker) { _taskQueue.Enqueue(task); } // 通知工作線程開始幹活 _waitHandle.Set(); } private void Work() { while (true) { string task = null; lock (_locker) { if (_taskQueue.Count > 0) { task = _taskQueue.Dequeue(); if (task == null) return; } } if (task != null) { Thread.Sleep(100); Console.WriteLine($"正在處理任務 {task}"); } else { // 如果任務等於空則阻塞線程,等待心的工作項 _waitHandle.WaitOne(); } } } public void Dispose() { // 優雅退出 EnqueueTask(null); _workThread.Join(); _waitHandle.Close(); } }
.NET 4.0 以後提供了一個
BlockingCollection<T>
類型實現了生產/消費者隊列。
4.2 ManualResetEvent
與
AutoResetEvent
類似,但在調用Set()
方法的時候打開門,是可以允許任意數量的線程在調用WaitOne()
後通過。(與AutoResetEvent
每次只能通過 1 個不一樣)如果是在關閉狀態下調用
WaitOne()
方法,線程會被阻塞,其余功能都與AutoResetEvent
一致。ManualResetEvent
的基類也是EventWaitHandle
,通過以下兩種方式均可構造。var manual1 = new ManualResetEvent(false); var manual2 = new EventWaitHandle(false, EventResetModel.ManualReset);
.NET 4.0 提供了性能更高的
ManualResetEventSliam
,但是不能夠跨線程使用。
4.3 CountdownEvent
使用
CountdownEvent
可以指定一個計數器的值,用於表明需要等待的線程數量。調用
Signal()
方法會將計數器自減 1 ,如果調用其Wait()
則會阻塞計數到 0 ,通過AddCount()
可以增加計數。class Program { static void Main() { var test = new CountdownEventTest(); new Thread(test.Say).Start("Hello 1"); new Thread(test.Say).Start("Hello 2"); new Thread(test.Say).Start("Hello 3"); test.CountdownEvent.Wait(); Console.WriteLine("所有線程執行完成..."); } } public class CountdownEventTest { public readonly CountdownEvent CountdownEvent = new CountdownEvent(3); public void Say(object info) { Thread.Sleep(1000); Console.WriteLine(info); CountdownEvent.Signal(); } }
當計數為 0 的時候,無法通過
AddCount()
增加計數,只能調用Reset()
進行復位。
4.4 等待句柄與線程池
除了手動開啟線程之外,事件等待句柄也支持通過線程池來運行工作任務。
通過
ThreadPool.RegisterWaitForSingleObject()
方法可以減少資源消耗,當需要執行的委托處於等待狀態的時候,不會浪費線程資源。class Program { static void Main() { var test = new ThreadPoolTest(); test.Test(); } } public class ThreadPoolTest { private readonly EventWaitHandle _waitHandle = new ManualResetEvent(false); public void Test() { RegisteredWaitHandle regHandle = ThreadPool.RegisterWaitForSingleObject(_waitHandle, Work, "OJBK", -1, true); Thread.Sleep(1000); _waitHandle.Set(); Console.ReadLine(); regHandle.Unregister(_waitHandle); } public void Work(object data,bool timeout) { Console.WriteLine($"正在執行任務 {data} ....."); } }
上述代碼如果通過傳統的方式進行阻塞與信號發送, 那麽有 1000 個請求
Work()
方法,就會造成大量服務線程阻塞,而RegisterWaitForSingleObject
可以立即返回,不會浪費線程資源。
4.5 跨進程的 EventWaitHandle
可以通過對 EventWaitHandle
類型構造函數的第三個參數傳入標識,來獲得跨進程的事件的等待句柄。
EventWaitHandle wh = new EventWaitHandle(false,EventResetMode.AutoReset,"AppName.Identity");
五、同步上下文
5.1 使用同步上下文
- 這裏的同步上下文不是
SynchronizationContext
類,而是 CLR 的自動鎖機制。 - 通過繼承
ContextBoundObject
基類並添加Synchronization
特性即可讓 CLR 自動加鎖。 - 同步上下文是一種比較重型的加鎖方法,很容易引起死鎖的情況發生。
5.2 重入
- 線程安全方法也被稱之為可重入的,因為其可以在運行途中被其他線程搶占。
- 使用了自動鎖會有一個嚴重問題,如果將
Synchronization
特性的reentrant
參數設置為true
。則允許同步類是可被重入的,這就導致同步上下文被臨時釋放,會導致過度期間任何線程都可以自由調用原對象的任何方法。 - 這是因為
Synchronization
特性是直接作用於類,所以其所有方法都會帶來可重入的問題。 - 所以因盡量減少粗粒度的自動鎖。
C# 多線程學習筆記 - 2