C#多執行緒之高階篇（上）

前言

拋開死鎖不談，只聊效能問題，儘管鎖總能粗暴的滿足同步需求，但一旦存在競爭關係，意味著一定會有執行緒被阻塞，競爭越激烈，被阻塞的執行緒越多，上下文切換次數越多，排程成本越大，顯然在高並行的場景下會損害效能。在高並行高效能且要求執行緒安全的述求下，無鎖構造（非阻塞構造）閃亮登場。

參考檔案：

C# - 理論與實踐中的 C# 記憶體模型，第 2 部分 | Microsoft Docs

volatile 關鍵字 (C# 參考)

一、非阻塞同步

重排序與快取

我們觀察下面這個例子：

public class Foo
{
    private int _answer;
    private bool _complete;

    void A() //A 1
    {
        _answer = 10;
        _complete = true;
    }

    void B() //B 2
    {
        if (_complete) Console.WriteLine(_answer);
    }
}

如果方法A和B在不同的執行緒上並行執行，B可能會列印 「 0 「 嗎？答案是會的，原因如下：

• 編譯器、CLR 或 CPU 可能會對程式碼/指令進行重排序（reorder）以提高效率。
• 編譯器、CLR 或 CPU 可能會進行快取優化，導致其它執行緒不能馬上看到變數的新值。

請務必重視它們，它們將是幽靈般的存在

int x = 0, y = 0, a = 0, b = 0;

var task1 = Task.Run(() => // A 1
{
    a = 1; // 1
    x = b; // 2
});
var task2 = Task.Run(() => // B 2
{
    b = 2; // 3
    y = a; // 4
});
Task.WaitAll(task1, task2);
Console.WriteLine("x:" + x + " y:" + y);

直覺和經驗告訴我們，程式至頂向下執行：程式碼1一定發生在程式碼2之前，程式碼3一定發生在程式碼4之前，然鵝

在一個獨立的執行緒中，每一個語句的執行順序是可以被保證的，但在不使用lock，waithandle這樣的顯式同步操作時，我們就沒法保證事件在不同的執行緒中看到的執行順序是一致的了。儘管執行緒A中一定需要觀察到a=1執行成功之後才會去執行x=b，但它沒法確保自己觀察得到執行緒B中對b的寫入，所以A還可能會列印出y的一箇舊版的值。這就叫指令重排序。

x:0 y:1 #1-2-3-4
x:2 y:0 #3-4-1-2
x:2 y:1 #1-3-2-4

可實際執行時還是有些讓我們驚訝的情況：

x:0 y:0 #??

這就是快取問題，如果兩個執行緒在不同的CPU上執行，每一個核心有自己的快取，這樣一個執行緒的寫入對於其它執行緒，在主記憶體同步之前就是不可見的了。

C#編譯器和CLR執行時會非常小心的保證上述優化不會破壞普通的單執行緒程式碼，和正確使用鎖的多執行緒程式碼。但有時，你仍然需要通過顯示的建立記憶體屏障（memory barrier，也稱作記憶體柵欄 （memory fence））來對抗這些優化，限制指令重排序和讀寫快取產生的影響。

記憶體屏障

參考部落格小林野夫

處理器支援哪種記憶體重排序（LoadLoad重排序、LoadStore重排序、StoreStore重排序、StoreLoad重排序），就會提供相對應能夠禁止重排序的指令，而這些指令就被稱之為記憶體屏障（LoadLoad屏障、LoadStore屏障、StoreStore屏障、StoreLoad屏障）

屏障名稱	範例	具體作用
StoreLoad	Store1;Store2;Store3;StoreLoad;Load1;Load2;Load3	禁止StoreLoad重排序，確保屏障之前任何一個寫（如Store2）的結果都會在屏障後任意一個讀操作（如Load1）載入之前被寫入
StoreStore	Store1;Store2;Store3;StoreStore;Store4;Store5;Store6	禁止StoreStore重排序，確保屏障之前任何一個寫（如Store1）的結果都會在屏障後任意一個寫操作（如Store4）之前被寫入
LoadLoad	Load1;Load2;Load3;LoadLoad;Load4;Load5;Load6	禁止LoadLoad重排序，確保屏障之前任何一個讀（如Load1）的資料都會在屏障後任意一個讀操作（如Load4）之前被載入
LoadStore	Load1;Load2;Load3;LoadStore;Store1;Store2;Store3	禁止LoadStore重排序，確保屏障之前任何一個讀（如Load1）的資料都會在屏障後任意一個寫操作（如Store1）的結果被寫入快取記憶體（或主記憶體）前被載入

讀屏障告訴處理器在執行任何的載入前，執行所有已經在失效佇列（Invalidte Queues）中的失效（I）指令。即：所有load barrier之前的store指令對之後（本核心和其他核心）的指令都是可見的。

Store Memory Barrier：寫屏障，等同於前文的StoreStore Barriers 將store buffer都寫入快取。

寫屏障告訴處理器在執行這之後的指令之前，執行所有已經在儲存快取（store buffer）中的修改（M）指令。即：所有store barrier之前的修改（M）指令都是對之後的指令可見。

最簡單的記憶體屏障是完全記憶體屏障（full memory barrier，或全柵欄（full fence）），它可以阻止所有跨越柵欄的指令進行重排並提交修改和重新整理快取。記憶體屏障之前的所有寫操作都要寫入記憶體，並將記憶體中的新值刷到快取，使得其它CPU核心能夠讀取到最新值，完全保證了資料的強一致性，進而解決CPU快取帶來的可見性問題。

我們簡單修改一下前面的案例

void A()
{
    _answer = 10;
    Thread.MemoryBarrier(); // 1
    _complete = true;
    Thread.MemoryBarrier(); // 3
}
void B()
{
    Thread.MemoryBarrier(); // 2
    if (_complete)
    {
        _testOutputHelper.WriteLine(_answer.ToString());
    }
}

屏障1,3使得這個例子不可能列印出0，屏障2保證如果B在A之後執行，_complete一定讀到的是true

記憶體屏障離我們並不遙遠，以下方式都會隱式的使用全柵欄：

• lock語法糖或Monitor.Enter / Monitor.Exit

• Interlocked類中的所有方法

• 使用執行緒池的非同步回撥，包括非同步委託，APM回撥，以及任務延續（task continuations）

• 訊號構造的等待/復位

• 任何依賴訊號同步的情況，比如啟動或等待Task，因此下面的程式碼也是執行緒安全的

  int x = 0;
  Task t = Task.Factory.StartNew (() => x++);
  t.Wait();
  Console.WriteLine (x);    // 1
  

volatile

另一個（更高階的）解決這個問題的方法是對_complete欄位使用volatile關鍵字。

volatile bool _complete;

volatile關鍵字通知編譯器在每個讀這個欄位的地方使用一個讀柵欄（acquire-fence），並且在每個寫這個欄位的地方使用一個寫柵欄（release-fence）。

這種「半柵欄（half-fences）」比全柵欄更快，因為它給了執行時和硬體更大的優化空間。

讀柵欄：也就是讀屏障（Store Memory Barrier），等同於前文的LoadLoad Barriers 將Invalidate的 都執行完成。告訴處理器在執行任何的載入前，執行所有已經在失效佇列（Invalidte Queues）中的失效（I）指令。即：所有load barrier之前的store指令對之後（本核心和其他核心）的指令都是可見的。

寫柵欄：也就是寫屏障（Store Memory Barrier），等同於前文的StoreStore Barriers 將store buffer都寫入主記憶體。
告訴處理器在執行這之後的指令之前，執行所有已經在儲存快取（store buffer）中的修改（M）指令。即：所有store barrier之前的修改（M）指令都是對之後的指令可見。

巧的是，Intel 的 X86 和 X64 處理器總是在讀時使用讀柵欄，寫時使用寫柵欄，無論是否使用volatile關鍵字。所以在使用這些處理器的情況下，這個關鍵字對硬體來說是無效的。然而，volatile關鍵字對編譯器和 CLR 進行的優化是有作用的，以及在 64 位 AMD 和 Itanium 處理器上也是有作用的。這意味著不能因為你的使用者端執行在特定型別的 CPU 上而放鬆警惕。

注意：使用volatile不能阻止寫-讀被交換

第一條指令	第二條指令	是否會被交換
讀	讀	不會
讀	寫	不會
寫	寫	不會（CLR 確保寫-寫操作永遠不會被交換，就算是沒有volatile關鍵字）
寫	讀	會！

在下面案例中仍然有可能會列印00的情況（對a的讀取可能發生在寫入前--重排序）

int a = 0, b = 0;
int x = 0, y = 0;
var task1 = Task.Run(() =>
{
    Thread.VolatileWrite(ref a, 1);
    x = Thread.VolatileRead(ref b);
});
var task2 = Task.Run(() =>
{
    Thread.VolatileWrite(ref b, 2);
    y = Thread.VolatileRead(ref a);
});
Task.WaitAll(task1, task2);

Console.WriteLine("x:" + x + " y:" + y);

volatile關鍵字不能應用於陣列元素，不能用在捕獲的區域性變數：這些情況下你必須使用VolatileRead和VolatileWrite方法

從上面的例子我們可以看出，寫-讀操作可能被重新排序，官方的解釋是：

在多處理器系統上，易失性讀取操作不保證獲取由任何處理器寫入該記憶體位置的最新值。 同樣，易失性寫入操作不保證寫入的值會立即對其他處理器可見。

（我的理解是：volatile關鍵字只能解決重排序問題，解決不了多處理器的快取一致性問題）

注意double 和 long無法標記為 volatile，因為對這些型別的欄位的讀取和寫入不能保證是原子的。 若要保護對這些型別欄位的多執行緒存取，請使用 Interlocked 類成員或使用 lock 語句保護存取許可權。

Interlocked

位於System.Threading，為多個執行緒共用的變數提供原子操作，這也是DOTNET為數不多的執行緒安全型別之一。

Interlocked通過將原子性的需求傳達給作業系統和CLR來進行實現其功能，此類的成員不會引發異常。

可以防止 1.執行緒上下文切換，2.執行緒更新可由其他執行緒存取的變數時，或者當兩個執行緒同時在不同的處理器上執行時 可能會出現的錯誤。

場景：

int i = 0;
i ++;

在大多數計算機上，自增並不是原子操作，需要以下步驟：

1. 將變數i的值載入到暫存器中。
2. 計算i + 1。
3. 將上面的計算結果儲存在變數i中。

假設A執行緒執行完1-2時被搶佔，B執行緒執行1-2-3，當A執行緒恢復時繼續執行3，此時B執行緒的值就被覆蓋掉了。

使用Increment即可解決，123會被打包成一個操作，以原子的方式實現自增

CAS

定義(摘自百度百科)：

CAS 操作包含三個運算元 —— 記憶體位置（V）、預期原值（A）和新值(B)。如果記憶體位置的值與預期原值相匹配，那麼處理器會自動將該位置值更新為新值。否則，處理器不做任何操作。無論哪種情況，它都會在 CAS 指令之前返回該位置的值。（在 CAS 的一些特殊情況下將僅返回 CAS 是否成功，而不提取當前值。）CAS 有效地說明了「我認為位置 V 應該包含值 A；如果包含該值，則將 B 放到這個位置；否則，不要更改該位置，只告訴我這個位置的值即可。」

Interlocked.CompareExchange，實現了CAS：比較兩個值是否相等，如果相等，則替換第一個值，否則什麼都不做，最終返回這個位置的原始值。

Interlocked.CompareExchange(ref _num, 1000, 500);

CAS在保證原子性讀寫的同時，沒有加鎖，保障了程式並行度，但也存在缺陷：

• ABA問題
• 只能保證一個地址的讀寫原子性
• 自旋CAS時間過長，容易給CPU帶來大開銷

二、延遲初始化

面試時候經常問：單例模式中的懶漢模式執行緒安全問題

場景：某個欄位構造開銷非常大，使得在初始化A時需要承擔初始化Expensive的開銷，即使Expensive欄位不會被用到。

public class A
{
    public readonly Expensive Expensive = new Expensive();
    // ..
}

public class Expensive
{
    // 構造開銷非常昂貴
}

自然會想到懶漢模式：按需載入

public class B
{
    private Expensive _expensive;

    public Expensive GetExpensiveInstance()
    {
        if (_expensive == null) _expensive = new Expensive();

        return _expensive;
    }
}

新的問題產生：GetExpensiveInstance是執行緒安全的嗎？我們可以通過加鎖解決

public class C
{
    private readonly object _locker = new object();
    private Expensive _expensive;

    public Expensive GetExpensiveInstance()
    {
        lock (_locker)
        {
            if (_expensive == null) _expensive = new Expensive();
            return _expensive;
        }
    }
}

現在面試官繼續問：還有效能更好的版本嗎？..

Lazy

net standard1.0 提供System.Lazy<T>來幫助你以執行緒安全且高效的方式（DCL）解決延遲初始化問題，只需

public class D
{
    private Lazy<Expensive> _expensive = new Lazy<Expensive>(() => new Expensive(), true);

    public Expensive GetExpensiveInstance() => _expensive.Value;
}

第一個引數是一個委託，告知如何構建，第二個引數是boolean型別，傳false實現的就是上面提到的plain B非執行緒安全遲初始化

雙檢鎖 double checked locking會進行一次額外的易失讀（volatile read），在物件已經完成初始化時，能夠避免獲取鎖產生的開銷。

public class E
{
    private readonly object _locker = new object();
    private volatile Expensive _expensive;

    public Expensive GetExpensiveInstance()
    {
        // 額外的易失讀（volatile read）
        if (_expensive == null)
        {
            lock (_locker)
            {
                if (_expensive == null) _expensive = new Expensive();
            }
        }
        
        return _expensive;
    }
}

LazyInitializer

LazyInitializer是一個靜態類，提供EnsureInitialized方法，第一個引數是需要構造的變數地址，第二個引數是一個委託，告知如何構造

public class F
{
    private Expensive _expensive;

    public Expensive GetExpensiveInstance()
    {
        LazyInitializer.EnsureInitialized(ref _expensive,
            () => new Expensive());
        return _expensive;
    }
}

它使用競爭初始化模式的實現，比雙檢鎖更快（在多核心情況下），因為它的實現完全不使用鎖。這是一個很少需要用到的極端優化，並且會帶來以下代價：

• 當參與初始化的執行緒數大於核心數時，它會更慢。
• 可能會因為進行了多餘的初始化而浪費 CPU 資源。
• 初始化邏輯必須是執行緒安全的（例如，Expensive的構造器對靜態欄位進行寫，就不是執行緒安全的）。
• 如果初始化的物件是需要進行銷燬的，多餘的物件需要額外的邏輯才能被銷燬。

競爭初始化（race-to-initialize）模式，通過易失性和CAS，實現無鎖構造

public class G
{
    private volatile Expensive _expensive;
    public Expensive Expensive
    {
        get
        {
            if (_expensive == null)
            {
                var instance = new Expensive();
                Interlocked.CompareExchange (ref _expensive, instance, null);
            }
            return _expensive;
        }
    }
}

三、執行緒區域性儲存

我們花費了大量篇幅來講並行存取公共資料問題，前文提到的鎖構造，訊號構造，無鎖構造本質上都是使用同步構造，使得多執行緒在存取公共資料時能安全的進行，然而有時我們會希望資料線上程間是隔離的，區域性變數就能實現這個目的，但他們的生命週期總是那麼短暫（隨程式碼塊而釋放），我們期待更大作用域的隔離資料，執行緒區域性變數（thread-local storage，TLS）就可以實現這個目的。

ThreadStatic

被ThreadStatic標記的static欄位不會線上程間共用，每個執行執行緒都有一個單獨的欄位範例

Note：

• 被標記的必須是static欄位，不能在範例欄位上使用（新增了也無效）
• 請不要給被標記的欄位指定初始值，因為這種初始化只會在類被構造時執行一次，影響一個執行緒，因此他依賴零值

如果你需要使用範例欄位，或者非零值，請使用ThreadLocal<T>

public class ThreadStatic測試
{
    private readonly ITestOutputHelper _testOutputHelper;
    [ThreadStatic] private static int _num;

    public ThreadStatic測試(ITestOutputHelper testOutputHelper)
    {
        _testOutputHelper = testOutputHelper;
    }

    [Fact]
    void Show()
    {
        void Work()
        {
            for (int i = 0; i < 100000; i++)
            {
                _num++;
                _testOutputHelper.WriteLine(_num.ToString());
            }
        }

        var t1 = new Thread(Work);
        var t2 = new Thread(Work);

        t1.Start();
        t2.Start();
        t1.Join();
        t2.Join();

        _testOutputHelper.WriteLine(_num.ToString());
    }
}

輸出：

100000
100000
0

LocalDataStoreSlot

封裝記憶體槽以儲存本地資料。 此類不能被繼承。.NET Framework 1.1加入，但在standard2.0+才有。

public sealed class LocalDataStoreSlot

.NET Framework 提供了兩種機制，用於使用執行緒本地儲存 (TLS) ：LocalDataStoreSlot和ThreadStaticAttribute

LocalDataStoreSlot比ThreadStaticAttribute更慢，更尷尬。此外，資料儲存為型別 Object，因此必須先將其強制轉換為正確的型別，然後再使用它。

有關使用 TLS 的詳細資訊，請參閱 執行緒本地儲存。

同樣，.NET Framework 提供了兩種使用上下文字地儲存的機制：LocalDataStoreSlot和ContextStaticAttribute。 上下文相對靜態欄位是用屬性標記的 ContextStaticAttribute 靜態欄位。 請參考註解

// 同一個 LocalDataStoreSlot 物件可以跨執行緒使用。
LocalDataStoreSlot _slot = Thread.AllocateNamedDataSlot("mySlot");
void Work()
{
    for (int i = 0; i < 100000; i++)
    {
        int num = (int)(Thread.GetData(_slot)??0);
        Thread.SetData(_slot, num + 1);
    }
    _testOutputHelper.WriteLine(((int)(Thread.GetData(_slot)??0)).ToString());
}
var t1 = new Thread(Work);
var t2 = new Thread(Work);
t1.Start();
t2.Start();
t1.Join();
t2.Join();
_testOutputHelper.WriteLine(((int)(Thread.GetData(_slot)??0)).ToString());

輸出效果和ThreadStaticAttribute一樣：

100000
100000
0

使用Thread.FreeNamedDataSlot("mySlot");可以釋放所有執行緒上的指定槽，但是隻有在所有對該槽的參照都出了其作用域，並且被垃圾回收後才會真正釋放。這確保了只要保持對槽的參照，就能繼續使用槽。

你也可以通過Thread.AllocateDataSlot()來建立一個無名槽位，與命名槽的區別是無名槽需要自行控制作用域

當然我們也可以對上面複雜的᠍᠍᠍᠍᠍Thread.GetData,Thread.SetData進行封裝

LocalDataStoreSlot _secSlot = Thread.GetNamedDataSlot ("securityLevel");
int Num
{
    get
    {
        object data = Thread.GetData(_secSlot);
        return data == null ? 0 : (int) data;    // null 相當於未初始化。
    }
    set { Thread.SetData (_secSlot, value); }
}

ThreadLocal

ThreadLocal<T>是 Framework 4.0 加入的，涵蓋在netstandard1.0。它提供了可用於靜態欄位和範例欄位的執行緒區域性儲存，並且允許設定預設值。

public class ThreadLocal測試
{
    ThreadLocal<int> _num = new ThreadLocal<int> (() => 3);
    private readonly ITestOutputHelper _testOutputHelper;


    public ThreadLocal測試(ITestOutputHelper testOutputHelper)
    {
        _testOutputHelper = testOutputHelper;
    }

    [Fact]
    void Show()
    {
        void Work()
        {
            for (int i = 0; i < 100000; i++)
            {
                _num.Value++;
            }
            _testOutputHelper.WriteLine(_num.ToString());
        }

        var t1 = new Thread(Work);
        var t2 = new Thread(Work);

        t1.Start();
        t2.Start();
        t1.Join();
        t2.Join();

        _testOutputHelper.WriteLine(_num.ToString());
    }
}

輸出

100003
100003
3

下面這個測試非常有意思

[Fact]
void Show()
{
    var threadName = new ThreadLocal<string>(() => "Thread" + Thread.CurrentThread.ManagedThreadId);
    Parallel.For(0, 13, x =>
    {
        bool repeat = threadName.IsValueCreated;
        _testOutputHelper.WriteLine($"ThreadName = {threadName.Value} {(repeat ? "(repeat)" : "")}");
    });
    
    threadName.Dispose();  // 釋放資源
}

你會發現當Parallel.For第二個引數超過你的邏輯核心後，repeat出現了！

ThreadName = Thread5 
ThreadName = Thread8 
ThreadName = Thread31 
ThreadName = Thread29 
ThreadName = Thread31 (repeat)
ThreadName = Thread30 
ThreadName = Thread18 
ThreadName = Thread12 
ThreadName = Thread32 
ThreadName = Thread28 
ThreadName = Thread33 
ThreadName = Thread35 
ThreadName = Thread34

Random類不是執行緒安全的，所以我們要不然在使用Random時加鎖（這樣限制了並行），如今我們有了ThreadLocal：

var localRandom = new ThreadLocal<Random>(() => new Random());

很輕易的就解決了執行緒安全問題，但是上面的版本使用的Random的無參構造方法，會依賴系統時間作為生成亂數的種子，在大概 10ms 時間內建立的兩個Random物件可能會使用相同的種子，下邊是解決這個問題的一個辦法：

var localRandom = new ThreadLocal<Random>(() => new Random (Guid.NewGuid().GetHashCode()) );

特別注意，不要以為GUID全域性唯一，GUID的HashCode也全域性唯一，上面的亂數仍然不是真隨機