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也全局唯一，上面的隨機數仍然不是真隨機

Tags:

VirMach 便宜 VPS

QNews

QNews