一、原子操作 先看一段問題代碼 /// <summary> /// 獲取自增 /// </summary> public static void GetIncrement() { long result = 0; Console.WriteLine("開始計算"); //10個併發執行 Parall ...
一、原子操作
先看一段問題代碼
/// <summary> /// 獲取自增 /// </summary> public static void GetIncrement() { long result = 0; Console.WriteLine("開始計算"); //10個併發執行 Parallel.For(0, 10, (i) => { for (int j = 0; j < 10000; j++) { result++; } }); Console.WriteLine("結束計算"); Console.WriteLine($"result正確值應為：{10000 * 10}"); Console.WriteLine($"result 現值為：{result}"); Console.ReadLine(); }
這是多線程下，result的值不同步的原因。
1.基於Lock實現
平時大家用的最多的應該就是加鎖了，同一時間，只有一個線程進入代碼塊。
實現代碼：
private static Object _obj = new object();
/// <summary> /// 原子操作基於Lock實現 /// </summary> public static void AtomicityForLock() { long result = 0; Console.WriteLine("開始計算"); //10個併發執行 Parallel.For(0, 10, (i) => { //lock鎖 lock (_obj) { for (int j = 0; j < 10000; j++) { result++; } } }); Console.WriteLine("結束計算"); Console.WriteLine($"result正確值應為：{10000 * 10}"); Console.WriteLine($"result 現值為：{result}"); Console.ReadLine(); }
結果：
2.基於CAS實現
CAS是一種有名的無鎖演算法。無鎖編程，即不適用鎖的情況下實現多線程之間的變數同步，也就是在沒有線程被阻塞的情況下實現變數的同步。
CAS在.NET中的實現類是Interlocked，內部提供很多原子操作的方法，最終都是調用Interlocked.CompareExchange(ref out,更新值，期望值) //基於記憶體屏障的方式操作 （七個步驟）
說到線程安全，不要一下子就想到加鎖，尤其是可能會調用頻繁或者是要求高性能的場合。
- CAS（Compare And Swap）比較並替換，是線程併發運行時用到的一種技術
- CAS是原子操作，保證併發安全，而不能保證併發同步
- CAS是CPU的一個指令（需要JNI調用Native方法，才能調用CPU的指令）
- CAS是非阻塞的、輕量級的樂觀鎖
CAS的適用場景
讀多寫少：如果有大量的寫操作，CPU開銷可能會過大，因為衝突失敗後會不斷重試（自旋），這個過程中會消耗CPU
單個變數原子操作：CAS機制所保證的只是一個變數的原子操作，而不能保證整個代碼塊的原子性，比如需要保證三個變數共同進行原子性的更新，就不得不使用悲觀鎖了
Interlocked主要函數如下：
Interlocked.Increment 原子操作，遞增指定變數的值並存儲結果。
Interlocked.Decrement 原子操作，遞減指定變數的值並存儲結果。
Interlocked.Add 原子操作，添加兩個整數並用兩者的和替換第一個整數
Interlocked.Exchange 原子操作，賦值
Interlocked.CompareExchange(ref a, b, c); 原子操作，a參數和c參數比較， 相等b替換a，不相等不替換。方法返回值始終是第一個參數的原值，也就是記憶體里的值
用Interlocked.Increment實現上面自增功能
代碼：
/// <summary> /// 自增CAS實現 /// </summary> public static void AtomicityForInterLock() { long result = 0; Console.WriteLine("開始計算"); Parallel.For(0, 10, (i) => { for (int j = 0; j < 10000; j++) { //自增 Interlocked.Increment(ref result); } }); Console.WriteLine($"結束計算"); Console.WriteLine($"result正確值應為：{10000 * 10}"); Console.WriteLine($"result 現值為：{result}"); Console.ReadLine(); }
結果：
Interlocked下原子操作的方法最終都是調用Interlocked.CompareExchange(ref a, b, c)實現的，現在我們利用CompareExchange自己實現一個原子操作功能
實現“一個變數自增到10000，然後又初始化到1開始自增的功能“
代碼：
/// <summary> /// 基於CAS原子操作自己寫 /// </summary> public static void AtomicityForMyCalc() { long result = 0; Console.WriteLine("開始計算"); Parallel.For(0, 10, (i) => { long init = 0; long incrementNum = 0; for (int j = 0; j < 10000; j++) { do { init = result; incrementNum = result + 1; incrementNum= incrementNum > 10000 ? 1 : incrementNum; //自增到10000後初始化成1 } //如果result=init,則result的值被incrementNum替換,否則result不變,返回的是result的原始值 while (init != Interlocked.CompareExchange(ref result, incrementNum, init)); if(incrementNum==10000) { Console.WriteLine($"自增到達10000啦!值被初始化為1"); } } }); Console.WriteLine($"結束計算"); Console.WriteLine($"result正確值應為：{10000}"); Console.WriteLine($"result 現值為：{result}"); Console.ReadLine(); }
結果：
3.自旋鎖SpinLock
自旋鎖（spinlock）：
是指當一個線程在獲取鎖的時候，如果鎖已經被其它線程獲取，那麼該線程將迴圈等待，然後不斷的判斷鎖是否能夠被成功獲取，直到獲取到鎖才會退出迴圈。
什麼情況下使用自旋鎖：
自旋鎖非常有助於避免阻塞，但是如果預期有大量阻塞，由於旋轉過多，您可能不應該使用自旋鎖。當鎖是細粒度的並且數量巨大（例如鏈接的列表中每個節點一個鎖）時以及鎖保持時間總是非常短時，旋轉可能非常有幫助。
短時間鎖定的情況下，自旋鎖（spinlock）更快。（因為自旋鎖本質上不會讓線程休眠，而是一直迴圈嘗試對資源訪問，直到可用。所以自旋鎖線程被阻塞時，不進行線程上下文切換，而是空轉等待。對於多核CPU而言，減少了切換線程上下文的開銷，從而提高了性能。）如果機器單核或鎖定時間長的要避免使用，因為占有著邏輯核心會導致其他的線程也不可用。
SpinLock和Lock的區別：
SpinLock，自旋鎖。嘗試獲取該鎖的線程持續不斷的check是否可以獲得。此時線程仍然是激活狀態，只是在空轉，浪費cpu而已。但是spinlock避免了線程調度和上下文切換，如果鎖的時間極短的話，使用該鎖反而效率會高。
而lock是線程被block了。這將引起線程調度和上下文切換等行為。
示例：
//創建自旋鎖 private static SpinLock spin = new SpinLock(); public static void Spinklock() { Action action = () => { bool lockTaken = false; try { //申請獲取鎖 spin.Enter(ref lockTaken); //臨界區 for (int i = 0; i < 10; i++) { Console.WriteLine($"當前線程{Thread.CurrentThread.ManagedThreadId.ToString()},輸出:1"); } } finally { //工作完畢，或者產生異常時，檢測一下當前線程是否占有鎖，如果有了鎖釋放它 //避免出行死鎖 if(lockTaken) { spin.Exit(); } } };
Action action2 = () => { bool lockTaken = false; try { //申請獲取鎖 spin.Enter(ref lockTaken); //臨界區 for (int i = 0; i < 10; i++) { Console.WriteLine($"當前線程{Thread.CurrentThread.ManagedThreadId.ToString()},輸出:2"); } } finally { //工作完畢，或者產生異常時，檢測一下當前線程是否占有鎖，如果有了鎖釋放它 //避免出行死鎖 if (lockTaken) { spin.Exit(); } } }; //並行執行2個action Parallel.Invoke(action, action2); }
結果：
申請鎖下麵的臨界區保證是順序執行的，不會因為多線程穿插輸出。
4.讀寫鎖ReaderWriterLockSlim
- 讀寫鎖是一個具有特殊用途的線程鎖，適用於頻繁讀取且讀取需要一定時間的場景，共用資源的讀取操作通常是可以同時執行的，
- 普通的互斥鎖不管是獲取還是修改操作無法同時執行，如果多個線程為了讀取操作而獲取互斥鎖，那麼同一時間只有一個線程可以執行讀取操作，
- 頻繁讀取的場景下會對吞吐量造成影響
- 讀寫鎖把鎖分為讀取鎖和寫入鎖，線程可以根據對共用資源的操作類型獲取讀取鎖還是寫入鎖，讀取鎖可以被多個線程同時獲取，寫入鎖不可以被多個線程同時獲取，且讀取鎖和寫入鎖不可以被不同的線同時獲取。
| 操作 | 讀取鎖狀態 | 寫入鎖狀態 | 獲取鎖是否需要等待 |
|---|---|---|---|
| 獲取讀取鎖 | 未被獲取 | 未被獲取 | 無需等待 |
| 獲取讀取鎖 | 已被其他線程獲取 | 未獲取 | 無需等待 |
| 獲取讀取鎖 | 未被獲取 | 已被其他線程獲取 | 需要等待其他線程釋放 |
| 獲取寫入鎖 | 未被獲取 | 未被獲取 | 無需等待 |
| 獲取寫入鎖 | 已被其他線程獲取 | 未被獲取 | 需要等待其他線程釋放 |
| 獲取寫入鎖 | 未被獲取 | 已被其他線程獲取 | 需要等待其他線程釋放 |
代碼示例：
//讀寫鎖， //策略支持遞歸 private static ReaderWriterLockSlim rwl = new ReaderWriterLockSlim(LockRecursionPolicy.SupportsRecursion); private static int index = 0; static void read() { try { //進入讀鎖 rwl.EnterReadLock(); for (int i = 0; i < 5; i++) { Console.WriteLine($"線程id:{Thread.CurrentThread.ManagedThreadId},讀數據,讀到index:{index}"); } } finally { //退出讀鎖 rwl.ExitReadLock(); } } static void write() { try { //嘗試獲寫鎖 while (!rwl.TryEnterWriteLock(50)) { Console.WriteLine($"線程id:{Thread.CurrentThread.ManagedThreadId},等待寫鎖"); } Console.WriteLine($"線程id:{Thread.CurrentThread.ManagedThreadId},獲取到寫鎖"); for (int i = 0; i < 5; i++) { index++; Thread.Sleep(50); } Console.WriteLine($"線程id:{Thread.CurrentThread.ManagedThreadId},寫操作完成"); } finally { //退出寫鎖 rwl.ExitWriteLock(); } } /// <summary> /// 執行多線程讀寫 /// </summary> public static void test() { var taskFactory = new TaskFactory(TaskCreationOptions.LongRunning, TaskContinuationOptions.None); Task[] task = new Task[6]; task[1] = taskFactory.StartNew(write); //寫 task[0] = taskFactory.StartNew(read); //讀 task[2] = taskFactory.StartNew(read); //讀 task[3] = taskFactory.StartNew(write); //寫 task[4] = taskFactory.StartNew(read); //讀 task[5] = taskFactory.StartNew(read); //讀 for (var i=0; i<6; i++) { task[i].Wait(); } }
可以看到，線上程4寫入期間，線程7是等待線程4寫萬再寫，線程7寫期間其它線程也沒有操場，等寫完後，讀的操作是多個線程交叉的。
適合的場景舉例：
- 多線程寫文件，多線程併發寫文件時，會報資源被占用錯誤，用這裡的寫鎖就可以獨占資源寫完再到下一個線程寫。
- 本地緩存的讀寫操作，幾個緩存值寫完才能讀出來，防止讀到不完整數據。
二、線程安全
1.線程安全集合
BlockingCollection：一個支持界限和阻塞的容器（線程安全集合），與隊列結構相似，常用函數如下
Add ：向容器中插入元素
TryTake：從容器中取出元素並刪除
TryPeek：從容器中取出元素，但不刪除。
CompleteAdding：告訴容器，添加元素完成。此時如果還想繼續添加會發生異常。
IsCompleted：告訴消費線程，生產者線程還在繼續運行中，任務還未完成。
普通用法示例：
/// <summary> /// 線程安全集合用法 /// </summary> public static void BC() { //線程安全集合 using (BlockingCollection<int> blocking = new BlockingCollection<int>()) { int NUMITEMS = 10000; for (int i = 1; i < NUMITEMS; i++) { blocking.Add(i); } //完成添加 blocking.CompleteAdding(); int outerSum = 0; // 定義一個委托方法取出集合元素 Action action = () => { int localItem; int localSum = 0; //取出並刪除元素，先進先出 while (blocking.TryTake(out localItem)) { localSum += localItem; } //兩數相加替換第一個值 Interlocked.Add(ref outerSum, localSum); }; //並行3個線程執行，多個線程同時取集合的數據 Parallel.Invoke(action, action, action); Console.WriteLine($"0+...{NUMITEMS-1} = {((NUMITEMS * (NUMITEMS - 1)) / 2)},輸出結果：{outerSum}"); //此集合是否已標記為已完成添加且為空 Console.WriteLine($"線程安全集合.IsCompleted={blocking.IsCompleted}"); } }
結果：
限制集合長度示例：
/// <summary> /// 限制集合長度 /// </summary> public static void BCLimtLength() { //限制集合長度為5個，後面進的會阻塞等集合少於5個再進來 BlockingCollection<int> blocking = new BlockingCollection<int>(5); var task1= Task.Run(() => { for (int i = 0; i < 20; i++) { blocking.Add(i); Console.WriteLine($"集合添加:({i})"); } blocking.CompleteAdding(); Console.WriteLine("完成添加"); }); // 延遲500ms執行等待先生產數據 var task2 = Task.Delay(500).ContinueWith((t) => { while (!blocking.IsCompleted) { var n = 0; if (blocking.TryTake(out n)) { Console.WriteLine($"取出:({n})"); } } Console.WriteLine("IsCompleted = true"); }); Task.WaitAll(task1, task2); }
結果：
在BlockingCollection中使用Stack(棧，先進後出)示例：
/// <summary> /// 線程安全集合，先進後出 /// </summary> public static void BCStack() { //線程安全集合，參數表明棧標識，隊列長度為5 BlockingCollection<int> blocking = new BlockingCollection<int>(new ConcurrentStack<int>(), 5); var task1 = Task.Run(() => { for (int i = 0; i < 20; i++) { blocking.Add(i); Console.WriteLine($"集合添加:({i})"); } blocking.CompleteAdding(); Console.WriteLine("完成添加"); }); // 等待先生產數據 var task2 = Task.Delay(500).ContinueWith((t) => { while (!blocking.IsCompleted) { var n = 0; if (blocking.TryTake(out n)) { Console.WriteLine($"取出:({n})"); } } Console.WriteLine("IsCompleted = true"); }); Task.WaitAll(task1, task2); }
一開始入了0-4，從最後的4開始取。
2.線程安全字典
ConcurrentDictionary :這個比較好理解，普通字典多線程併發時添加時會報錯，而這個則是線程安全的，不會報錯。
普通字典示例：
//普通字典 private static IDictionary<string, string> Dictionaries { get; set; } = new Dictionary<string, string>(); /// <summary> /// 字典增加值 /// </summary> public static void AddDictionaries() { Stopwatch sw = new Stopwatch(); sw.Start(); //併發1000個線程寫 Parallel.For(0, 1000, (i) => { var key = $"key-{i}"; var value = $"value-{i}"; // 不加鎖會報錯 // lock (Dictionaries) // { Dictionaries.Add(key, value); // } }); sw.Stop(); Console.WriteLine("Dictionaries 當前數據量為： {0}", Dictionaries.Count); Console.WriteLine("Dictionaries 執行時間為： {0} ms", sw.ElapsedMilliseconds); }
不加鎖時結果：
加鎖後：
線程安全字典示例：
//線程安全字典 private static IDictionary<string, string> ConcurrentDictionaries { get; set; } = new ConcurrentDictionary<string, string>(); /// <summary> /// 線程安全字典添加值 /// </summary> public static void AddConcurrentDictionaries() { Stopwatch sw = new Stopwatch(); sw.Start(); //併發1000個線程寫 Parallel.For(0, 1000, (i) => { var key = $"key-{i}"; var value = $"value-{i}"; // 無須加鎖 ConcurrentDictionaries.Add(key, value); }); sw.Stop(); Console.WriteLine("ConcurrentDictionaries 當前數據量為： {0}", ConcurrentDictionaries.Count); Console.WriteLine("ConcurrentDictionaries 執行時間為： {0} ms", sw.ElapsedMilliseconds); }
可以看到，線程安全字典比普通字典性能略好，且線程安全字典無需加鎖。
三、線程池
1.通過QueueUserWorkItem啟動工作者線程
ThreadPool線程池中有兩個重載的靜態方法可以直接啟動工作者線程：
- ThreadPool.QueueUserWorkItem(waitCallback)；
- ThreadPool.QueueUserWorkItem(waitCallback,Object)；
先把WaitCallback委托指向一個帶有Object參數的無返回值方法，再使用ThreadPool.QueueUserWorkItem(WaitCallback)就可以一步啟動此方法，此時非同步方法的參數被視為null。
示例1：
public class ThreadLoopDemo { /// <summary> /// 回調方法 /// </summary> /// <param name="obj"></param> static void CallMethod(object state) { Console.WriteLine("RunWorkerThread開始工作"); Console.WriteLine("工作者線程啟動成功!"); } public static void Test() { //工作者線程最大數目，I/O線程的最大數目 ThreadPool.SetMaxThreads(1000, 1000);
//啟動工作者線程 ThreadPool.QueueUserWorkItem(new WaitCallback(CallMethod!)); Console.ReadKey(); } }
執行Test方法，結果：
示例2：
使用第二個重載方法ThreadPool.QueueUserWorkItem(WaitCallback,object)方法可以把object對象作為參數傳送到回調函數中。
public class ThreadLoopDemo { /// <summary> /// 回調方法 /// </summary> /// <param name="obj"></param> static void CallMethod(object state) { Console.WriteLine("RunWorkerThread開始工作"); Order order=state as Order; Console.WriteLine($"orderName:{order.orderName},price:{order.price}"); Console.WriteLine("工作者線程啟動成功!"); } public static void Test() { //工作者線程最大數目，I/O線程的最大數目 ThreadPool.SetMaxThreads(1000, 1000); Order order = new Order() { orderName = "冰箱", price = 1888 }; //啟動工作者線程 ThreadPool.QueueUserWorkItem(new WaitCallback(CallMethod!),order); Console.ReadKey(); } } public class Order { public string orderName { get; set; } public decimal price { get; set; } }
執行Test方法，結果：
通過ThreadPool.QueueUserWork啟動工作者線程非常方便，但是WaitCallback委托指向的必須是一個帶有object參數的無返回值方法。
線程池還可以重用線程，比喻可以吧線程池大小設為5個，去執行一批任務，防止大量創建新線程損耗大量cpu。
所以這個方法啟動的工作者線程僅僅適合於帶單個參數和無返回值的情況。
2.線程池等待（信號量）
ThreadPool並沒有Thread的Join等待介面，那麼想讓ThreadPool等待要這麼做呢？
ManualResetEvent:通知一個或多個正在等待的線程已發生的事件，相當於發送了一個信號
mre.WaitOne:卡住當前主線程，一直等到信號修改為true的時候，才會接著往下跑
public class ThreadLoopDemo { /// <summary> /// 執行 /// </summary> public static void Test() { //用來控制線程等待,false預設為關閉狀態 ManualResetEvent mre = new ManualResetEvent(false); ThreadPool.QueueUserWorkItem(p => { Console.WriteLine("線程1開始
