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 ++;

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

將變數i的值加載到暫存器中，
計算i + 1，
將上面的計算結果存盤在變數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 = https://www.cnblogs.com/xiaolipro/p/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也全域唯一，上面的亂數仍然不是真隨機

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

標籤：C#

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

下一篇：C# 鎖匯總