C#多執行緒之同步基礎篇-有解無憂

一、基本概念
二、鎖構造
- - Monitor
  - Mutex
  - 死鎖
三、信號構造
- - Semaphore
  - ManualResetEvent
  - AutoResetEvent
  - CountdownEvent
四、等待句柄
- - 等待句柄和執行緒池
  - WaitHandle

一、基本概念

執行緒安全（thread safe）：指的是被任意多的執行緒同時執行，都可以保證正確性，

除基本型別外，很少有型別是執行緒安全的，執行緒安全的責任基本落在開發者身上，System.Collections.Concurrent命名空間下的型別的除外，

執行緒安全最常見的手段一般是使用【排它鎖】，將大段代碼甚至是訪問的整個物件封裝在一個排它鎖內，從而保證在高層上能進行順序訪問，
這種解決方案適用于物件的方法都能夠快速執行的場景（否則會導致大量的阻塞），
還有一種手段很高明，即通過【最小化共享資料】來減少【執行緒互動】，web服務器就是最好的案例，由于多個客戶端請求可以同時到達，服務端方法必須保證執行緒安全，類似的案例還有【無狀態】設計，在本質上限制了
資料互動的可能，具有良好的伸縮性（scalability），
還有一種手段，【自動鎖機制（automatic locking）】如果繼承 ContextBoundObject 類并使用 Synchronization 特性，.NET Framework 就可以實作這種機制，framework全系支持，但是netcore沒有，類似java的synchronized，

盡管這樣降低了開發者實作執行緒安全的負擔，但范圍過大的鎖定作用域將制造出巨大的麻煩：死鎖、非有意的重入以及降低并發度，這使得手動鎖定在任何場景都顯得更為合適，而并不僅僅只在簡單的場景下（直到有更好用的自動鎖機制出現），

其它手段，【信號構造】，【記憶體屏障】，【自旋構造】，，，

同步（synchronization）：指對在一個系統中所發生的事件（event）之間進行協調，在時間上出現一致性與統一化的現象 -- 為期望的結果協調多個執行緒的行為，

當多個執行緒訪問同一個資料時，同步尤為重要，但這是一件非常容易G的事情，

同步物件（synchronized object）：對所有參與同步的執行緒可見的任何物件都可以被當作同步物件使用，但有一個硬性規定：同步物件必須為參考型別，同步物件一般是私有的（因為這有助于封裝鎖邏輯），

同步物件也可以就是其要保護的物件，

class ThreadSafe
{
  List <string> _list = new List <string>();

  void Test()
  {
    lock (_list)
    {
      _list.Add ("Item 1");
      // ...

一個只被用來加鎖的欄位可以精確控制鎖的作用域與粒度，

private static readonly object _locker = new object();

物件自己（this），甚至是型別，lambda 運算式或匿名方法所捕獲的區域變數都可以被當作同步物件來使用：

lock (this) { ... }
// 或者：
lock (typeof (Widget)) { ... }    // 保護對靜態資源的訪問

但這種方式的缺點在于并沒有對鎖邏輯進行封裝，從而很難避免【死鎖】或過多的【阻塞】，同時型別上的鎖也可能會跨越應用程式域（application domain）邊界（在同一行程內），

阻塞（block）：當執行緒的執行由于某些原因被暫停，比如【信號構造】或【鎖構造】時，比如呼叫Thread.Sleep，Task.Wait，或者通過Join方法等待其它執行緒結束時，則認為此執行緒被阻塞（blocked），

阻塞會在以下 4 種情況下解除（電源按鈕可不能算╮(╯▽╰)╭）：

阻塞條件被滿足
操作超時（如果指定了超時時間）
通過Thread.Interrupt中斷
通過Thread.Abort中止

編譯器將async Task轉換為狀態機，到達 await 時暫停執行等待后臺作業完成時繼續執行，從理論上講，這是異步的承諾模型的實作，）

static async Task<Toast> MakeToastWithButterAndJamAsync(int number)
{
    // do samething..
    var toast = await ToastBreadAsync(number);
    // do samething..
    return toast;
}

鎖構造（lock）：鎖能夠限制同一時刻可以執行某些指令或是某段代碼的執行緒數量，排他鎖是最常見的，它只允許同一時刻至多有一個執行緒執行，從而可以使得參與競爭的執行緒在訪問公共資料時不會彼此干擾，
一般的排他鎖有lock（Monitor.Enter/Monitor.Exit）、Mutex、SpinLock，非排他鎖有Semaphore、SemaphoreSlim以及reader/writer lock，

信號構造（signal）：信號構造可以使一個執行緒【掛起】，直到接收到另一個執行緒的通知，避免了低效的輪詢，有兩種經常使用的信號設施：事件等待句柄（event wait handle ）和Monitor類的Wait / Pulse方法，
Framework 4.0 加入了CountdownEvent與Barrier類，

自旋（spinning）：有時執行緒必須阻塞/暫停，直至條件被滿足，【信號構造】或【鎖構造】可以實作，但在等待條件能夠在微秒級的時間被滿足時，
自旋往往更加高效，因為它避免了背景關系切換帶來的昂貴開銷，

while (!condition);

自旋往往與阻塞組合使用，防止cpu浪費

while (!condition) Thread.Sleep (10);

執行緒狀態（thread state）：Unstarted、Running、WaitSleepJoin、Stopped，，

指令原子性（instruction atomically）：如果一組【指令】可以在 CPU 上不可分割地執行，那么它就是原子的

原子性（atomically）：如果一組變數總是在相同的鎖內進行讀寫，就可以稱為原子性讀寫

lock (locker) { if (x != 0) y /= x; }

可以說x和y是被原子的訪問的，因為上面的代碼塊無法被其它的執行緒分割或搶占（cpu悲觀鎖/總線鎖），如果被其它執行緒分割或搶占，x和y就可能被別的執行緒修改導致計算結果無效（cpu樂觀鎖/快取鎖），而現在 x和y總是在相同的排它鎖中進行訪問，因此不會出現除數為零的錯誤，

如果lock代碼塊內發生例外，原子性將被打破

decimal _savingsBalance, _checkBalance;

void Transfer (decimal amount)
{
  lock (_locker)
  {
    _savingsBalance += amount;
    _checkBalance -= amount + GetBankFee();
  }
}

如果GetBankFee()方法內拋出例外，銀行可能就要虧錢了，在這個例子中，我們可以通過更早的呼叫GetBankFee()來避免這個問題，對于更復雜情況，解決方案是在catch或finally中實作“回滾（rollback）”邏輯，

二、鎖構造

Monitor

C# 的lock陳述句是一個語法糖，它其實就是使用了try / finally來呼叫Monitor.Enter與Monitor.Exit方法：

bool taken = false;
try
{
    // JIT應該行內此方法，以便在典型情況下優化lockTaken引數的檢查，
    Monitor.Enter(_locker,ref taken);
    num++;
}
finally
{
    // C# 4.0 解決鎖泄露問題
    if (taken) Monitor.Exit(_locker);
}

Monitor是【可重入的（Reentrant）】,只有當最外層的lock陳述句退出或是執行了匹配數目的Monitor.Exit陳述句時，物件才會被解鎖，

static void Main()
{
    lock (locker)  // 執行緒只會在第一個（最外層）lock處阻塞，
    {
        AnotherMethod();
        // 這里依然擁有鎖，因為鎖是可重入的
    }
}

static void AnotherMethod()
{
  lock (_locker) { Console.WriteLine ("Another method"); }
}

Monitor的性能：在一個 2010 時代的計算機上，沒有競爭的情況下獲取并釋放鎖一般只需 20 納秒，如果存在競爭，產生的背景關系切換會把開銷增加到微秒的級別，并且執行緒被重新調度前可能還會等待更久的時間，如果需要鎖定的時間很短，那么可以使用【自旋鎖（SpinLock）】來避免背景關系切換的開銷，

Monitor還提供了一個TryEnter方法，允許以毫秒或是TimeSpan方式指定超時時間，如果獲得了鎖，該方法會回傳true，而如果由于超時沒有獲得鎖，則會回傳false，TryEnter也可以不傳遞超時時間進行呼叫，這是對鎖進行“測驗”，如果不能立即獲得鎖就會立即回傳false，

如果獲取鎖后保持的時間太長而不釋放，就會降低并發度，同時也會加大【死鎖】的風險，

Mutex

Mutex互斥體類似于Monitor，不同在于它是可以跨越行程作業，換句話說，Mutex可以是機器范圍（computer-wide）的，也可以是程式范圍（application-wide）的，

沒有競爭的情況下，獲取并釋放Mutex需要幾微秒的時間，大約比lock慢 50 倍，

使用Mutex類時，可以呼叫WaitOne方法來加鎖，呼叫ReleaseMutex方法來解鎖，關倍訓銷毀Mutex會自動釋放鎖，與lock陳述句一樣，Mutex只能被獲得該鎖的執行緒釋放，

跨行程Mutex的一種常見的應用就是確保只運行一個程式實體，

// 命名的 Mutex 是機器范圍的，它的名稱需要是唯一的
// 比如使用公司名+程式名，或者也可以用 URL
using (var mutex = new Mutex (false, "Global\oreilly.com OneAtATimeDemo"))
{
  // 可能其它程式實體正在關閉，所以可以等待幾秒來讓其它實體完成關閉
  if (!mutex.WaitOne (TimeSpan.FromSeconds (3), false))
  {
    Console.WriteLine ("Another app instance is running. Bye!");
    return;
  }
  RunProgram();
}

如果在終端服務（Terminal Services）下運行，機器范圍的Mutex默認僅對于運行在相同終端會話的應用程式可見，要使其對所有終端會話可見，需要在其名字前加上Global\，

死鎖

當兩個執行緒等待的資源都被對方占用時（A等B，B等A），它們都無法執行，這就產生了死鎖，更復雜的死鎖鏈可能由三個或更多的執行緒創建，

object locker1 = new object();
object locker2 = new object();

new Thread(() =>
{
    lock (locker1)
    {
        Thread.Sleep(1000);
        lock (locker2)  // 死鎖
        {
            // do something..
        }
    }
}).Start();

lock (locker2)
{
    Thread.Sleep(1000);
    lock (locker1)  // 死鎖
    {
        // do something..
    }
}

CLR 不會像SQL Server一樣自動檢測和解決死鎖，除非你指定了鎖定的超時時間，否則死鎖會造成參與的執行緒無限阻塞，（在SQL CLR 集成宿主環境中，死鎖能夠被自動檢測，并在其中一個執行緒上拋出可捕獲的例外，）

死鎖是多執行緒中最難解決的問題之一，尤其是在有很多關聯物件的時候，這個困難在根本上在于無法確定呼叫方（caller）已經擁有了哪些鎖，

你可能會鎖定類x中的私有欄位a，而并不知道呼叫方（或者呼叫方的呼叫方）已經鎖住了類y中的欄位b，同時，另一個執行緒正在執行順序相反的操作，這樣就創建了死鎖，諷刺的是，這個問題會由于（良好的）面向物件的設計模式而加劇，因為這類模式建立的呼叫鏈直到運行時才能確定，

流行的建議：“以一致的順序對物件加鎖以避免死鎖”，盡管它對于我們最初的例子有幫助，但是很難應用到剛才所描述的場景，更好的策略是：如果發現在鎖區域中的對其它類的方法呼叫最侄訓參考回當前物件，就應該小心，同時考慮是否真的需要對其它類的方法呼叫加鎖（往往是需要的，但是有時也會有其它選擇），更多的依靠宣告方式（declarative）與資料并行（data parallelism）、不可變型別（immutable types）與非阻塞同步構造（ nonblocking synchronization constructs），可以減少對鎖的需要，

有另一種思路：當你在擁有鎖的情況下訪問其它類的代碼，對于鎖的封裝就存在潛在的泄露，這不是 CLR 或 .NET Framework 的問題，而是因為鎖本身的局限性，某些人認為造成這樣問題的根因是可重入？

三、信號構造

Semaphore

信號量類似于一個通道：它具有一定的容量（capacity）房間，并且有保安把守，一旦滿員，就不允許其他人進入，這些人將在外面排隊，當有一個人離開時，排在最前頭的人便可以進入，

容量為1的的信號量就是一把互斥鎖，類似mutex，不同的是信號量沒有【所有者】，它是執行緒無關（thread-agnostic）的，任何執行緒都可以在呼叫Semaphore上的Release方法，而對于mutex，只有獲得鎖的執行緒才可以釋放該鎖，

SemaphoreSlim是 standard1.0 就支持的輕量級的信號量，功能與Semaphore相似，不同之處是它對于并行編程的低延遲需求做了優化；支持在等待時指定取消標記（cancellation token），但它不能跨行程使用，Semaphore可以，

在Semaphore上呼叫WaitOne或Release會產生大概 1 微秒的開銷（無競爭情況下），而SemaphoreSlim產生的開銷約是其四分之一，

ManualResetEvent

ManualResetEvent 呼叫WaitOne進入阻塞，任意可訪問的執行緒都能呼叫Set方法來放行，

var waitHandle = new ManualResetEvent(false);
// var waitHandle = new EventWaitHandle(false, EventResetMode.ManualReset);
Task.Run(() =>
{
    _testOutputHelper.WriteLine(Thread.CurrentThread.ManagedThreadId + " 嘗試進門...");
    waitHandle.WaitOne();
    _testOutputHelper.WriteLine(Thread.CurrentThread.ManagedThreadId + " 進去了");
    業務邏輯();
    _testOutputHelper.WriteLine("當前門的狀態是開啟的嗎？"+waitHandle.WaitOne(0)); //true
});
Thread.Sleep(1000);
_testOutputHelper.WriteLine(Thread.CurrentThread.ManagedThreadId + " say：我來開門");
waitHandle.Set();

AutoResetEvent

AutoResetEvent 如其命名，收到通知后他能自動復位（reset），而ManualResetEvent不能，

EventWaitHandle waitHandle = new AutoResetEvent(false);
 // var waitHandle2 = new EventWaitHandle(false, EventResetMode.AutoReset);
 Task.Factory.StartNew(() =>
 {
     _testOutputHelper.WriteLine(Thread.CurrentThread.ManagedThreadId + " 嘗試進門...");
     waitHandle.WaitOne();
     _testOutputHelper.WriteLine(Thread.CurrentThread.ManagedThreadId + " 進去了");
     業務邏輯();
     _testOutputHelper.WriteLine("當前門的狀態是開啟的嗎？"+waitHandle.WaitOne(0)); // false
     waitHandle.Set();
     Task.Run(() =>
     {
         waitHandle.WaitOne();  // 永遠阻塞，直至主執行緒退出
         _testOutputHelper.WriteLine(Thread.CurrentThread.ManagedThreadId + " 進去了");
         業務邏輯();
     });
 });
 Thread.Sleep(1000);
 _testOutputHelper.WriteLine(Thread.CurrentThread.ManagedThreadId + " say：我來開門");
 waitHandle.Set();
 
 Thread.Sleep(1000); // 等待worker

從 Framework 4.0 開始，提供了另一個版本的ManualResetEvent，名為ManualResetEventSlim，后者為短等待時間做了優化，它提供了進行一定次數迭代自旋的能力，也實作了一種更有效的管理機制，允許通過CancellationToken取消Wait等待，但它不能用于跨行程的信號同步， ManualResetEventSlim不是WaitHandle的子類，但它提供一個WaitHandle的屬性，會回傳一個基于WaitHandle的物件（使用它的性能和一般的等待句柄相同），

EventWaitHandle的構造方法允許以命名的方式進行創建，這樣它就可以跨行程使用，名稱就是一個字串，可以隨意起名，但是注意不要和別人的命名沖突！如果名字在計算機上已存在，你就會獲取一個它對應的EventWaitHandle的參考，否則作業系統會創建一個新的，

EventWaitHandle wh = new EventWaitHandle (false,EventResetMode.AutoReset,"MyCompany.MyApp.SomeName");

使用AutoResetEvent實作簡易的生產消費佇列

public class PCQueue:IDisposable
{
    private EventWaitHandle _waitHandle = new AutoResetEvent(false);
    private Thread _worker;
    private readonly object _locker = new object();
    private Queue<string> _tasks = new Queue<string>();

    private readonly ITestOutputHelper _testOutputHelper;

    public PCQueue(ITestOutputHelper testOutputHelper)
    {
        _testOutputHelper = testOutputHelper;

        _worker = new Thread(Work);
        _worker.Start();
    }

    public void AddTask(string task)
    {
        lock (_locker)
        {
            _tasks.Enqueue(task);
        }

        _waitHandle.Set();
    }

    private void Work()
    {
        while (true)
        {
            string task = null;
            lock (_locker)
            {
                if (_tasks.Count > 0)
                {
                    task = _tasks.Dequeue();
                    if (task == null) return;  // null為退出任務
                }
            }

            if (task == null)
            {
                _waitHandle.WaitOne();  // 沒有任務，進入阻塞，等待新的任務
            }
            else
            {
                _testOutputHelper.WriteLine("執行任務：" +task);
                Thread.Sleep(1000);  // 模擬耗時任務
            }
        }
    }

    public void Dispose()
    {
        AddTask(null);  // 通知消費執行緒退出
        _worker.Join();  // 等待消費執行緒執行完成
        _waitHandle.Close();  // 釋放事件句柄
        _tasks.Clear();
    }
}

CountdownEvent

CountdownEvent可以用于多執行緒等待，這個型別是 Framework 4.0 加入的，并且是一個高效的純托管實作，

CountdownEvent _countdown = new CountdownEvent (3);

void 測驗CountdownEvent等待()
{
    Task.Run(作業);
    Task.Run(作業);
    Task.Run(作業);

    _countdown.Wait();
    _testOutputHelper.WriteLine("大家都干完了");
}

void 作業()
{
    _testOutputHelper.WriteLine("干活");
    Thread.Sleep(1000);
    _countdown.Signal();
}

四、等待句柄

等待句柄和執行緒池

如果你的應用有很多執行緒，這些執行緒大部分時間都在阻塞，那么可以通過呼叫ThreadPool.RegisterWaitForSingleObject來減少資源消耗，當向等待句柄發信號時（或者已超時），委托（這里的Work方法）會在一個執行緒池執行緒運行，

[Fact]
void Show()
{
    var _waitHandle = new ManualResetEvent(false);
    var reg = ThreadPool.RegisterWaitForSingleObject(_waitHandle, Work, "hahah", -1, true);
    Thread.Sleep(3000);
    _testOutputHelper.WriteLine("發送復位信號");
    _waitHandle.Set();
    reg.Unregister(_waitHandle);
}
private void Work (object data, bool timedOut)
{
    _testOutputHelper.WriteLine ("Say - " + data);
    // 執行任務 ....
}

WaitHandle

除了Set、WaitOne和Reset方法外，在WaitHandle類上還有一些靜態方法用來解決更復雜的同步問題，

WaitAny、WaitAll和SignalAndWait方法可以向多個等待句柄發信號和進行等待操作，等待句柄可以是不同的型別（包括 Mutex、Semaphore、CountdownEvent等，因為它們都派生自抽象類WaitHandle），

waitAny 等待一組等待句柄中任意一個
WaitAll 等待給定的所有等待句柄，這個等待是原子的，
SignalAndWait會呼叫一個等待句柄的Set方法，然后呼叫另一個等待句柄的WaitOne方法，在向第一個句柄發信號后，會（讓當前執行緒）跳到第二個句柄的等待佇列的最前位置（插隊），
```
WaitHandle.SignalAndWait (wh1, wh2);
```

轉載請註明出處，本文鏈接：https://www.uj5u.com/net/532480.html

標籤：.NET技术

上一篇：C#多執行緒之執行緒基礎篇

下一篇：學習ASP.NET Core Blazor編程系列十——路由（中）