CLR 混合線程同步構造-續

本篇繼上一篇討論一下多線程併發的處理狀況，以及如何編寫異步的同步構造代碼避免線程阻塞。CLR 到此篇就結束了，若是想看Jeffrey 原著的請留言，寫下郵箱地址。

 

著名的雙檢索技術

CLR 很好的支持雙檢索技術，這應該歸功於CLR 的內存模型以及 volatile 字段訪問，如下代碼演示瞭如何使用 C# 雙檢索技術。

public sealed class Singleton {
    private static Object s_lock = new Object();
    private static Singletion s_value = null;
​
    private Singleton()
   {
        //初始化單實例對象的代碼放在這裏 ...
   }
​
    public static Singleton GetSingleton()
   {
        if（s_value != null） return s_value;
​
        Monitor.Enter(s_lock);//尚未建立讓一個線程建立它
        if(s_value == null)
       {
            Singleton temp = new Singleton();
            Volatile.Write(ref s_value, temp);
       }
        Monitor.Exit(s_lock);
        return s_value;
   }
}

雙檢索技術背後的思路在於，對 GetSingletion 方法的一個調用能夠快速地檢查 s_value 字段，判斷對象是否建立。

在CLR 中，對任何鎖方法的調用都構成了一個完整的內存柵欄，在柵欄以前寫入的任何變量都必須在柵欄以前完成；在柵欄以後的任何變量讀取都必須在柵欄以後開始。對於GetSingleton 方法，這意味着 s_value 字段的值必須在調用了 Monitor.Enter 以後從新讀取。調用前緩存到寄存器中的東西做不了數。

 

假如第二個 if 語句中包含的是下面這行代碼：

s_value = new Singleton(); //你極有可能這樣寫

你覺得編譯器會生成代碼爲一個 Singleton 分配內存，調用構造器來初始化字段，再將引用賦給 s_value 字段。使一個指對其餘線程可見成爲 發佈。但這只是你一廂情願的想法。編譯器可能會這樣作：爲Singleton 分配內存，將引用發佈到 s_value，再調用構造器。

從單線程的角度看，像這樣改變順序是可有可無的。但再將引用發佈給 s_value 以後，並在調用構造器以前，若是另外一個線程調用了 GetSingleton 方法，那會發生什麼？這個線程會發現 s_value 不爲 null，因此會開始使用 Singleton 對象，但對象的構造器尚未結束執行呢！

大多數時候，這個技術會損害效率。下面沒有使用雙檢索技術，但行爲和上一個版本相同。還更簡潔了。

internal sealed class Singleton{
    private static Singleton s_value = new Singleton();
​
    //私有構造器防止 這個類外部的任何代碼建立一個實例
    private Singleton(){ 
        //初始化單實例對象的代碼放在這裏
   }
    public static Singleton GetSingleton() { return s_value; }
}

因爲代碼首次訪問類的成員時，CLR 會自動調用類型的類構造器，因此首次有一個線程查詢 Singleton 的 GetSingleton 方法時，CLR 就會自動調用類構造器，從而建立一個對象實例。

這種方式的缺點在於，首次訪問類的 任何成員 都會調用類型構造器。因此，若是Singleton 類型定義了其餘靜態成員，就會在訪問其餘任何靜態成員時建立 Singleton 對象。 有人經過定義嵌套類型來解決這個問題。

internal sealed class Singleton{
    private static Singleton s_value = null;
    //私有構造器阻止這個類外部的任何代碼建立實例
    private Singleton()
   {
        
   }
​
    //如下公共靜態方法返回單實例對象
    public static Singleton GetSingleton()
   {
        if(s_value != null) return s_value;
        //建立一個新的單實例對象，並把它固定下來（若是另外一個線程尚未固定它的話）
        Singleton temp = new Singleton();        
        Interlocked.CompareExchange(ref s_value, temp , null);
​
        //若是這個線程競爭失敗，新建的第二個單實例對象會被垃圾回收
        return s_value;
   }
}

因爲大多數應用都不多發生多個線程同時調用 GetSingleton 的狀況，因此不太可能同時建立多個Singleton 對象。

上述代碼有不少方面的優點，1 它的速度很是快。2 它永不阻塞任何線程。

若是，一個線程池線程在一個 Monitor 或者 其餘任何內核模式的線程同步構造上阻塞，線程池線程就會建立另外一個線程來保持 CPU 的「飽和」。所以會初始化更多的內存，而其全部 DLL 都會收到一個線程鏈接通知。

 

FCL 有兩個類型封裝了本節描述的模式。下面是泛型System.Lazy 類。

public class Lazy<T> {
    public Lazy(Func<T> valueFactory, LazyThreadSafetyMode mode);
    public Boolean IsValueCreated { get; }
    public T Value { get; }
}

下面代碼演示它如何工做的：

public static void Main() {
    // Create a lazy-initialization wrapper around getting the DateTime
    Lazy<String> s = new Lazy<String>(() => DateTime.Now.ToLongTimeString(), true);
    
    Console.WriteLine(s.IsValueCreated); // Returns false because Value not queried yet
    Console.WriteLine(s.Value); // The delegate is invoked now
    Console.WriteLine(s.IsValueCreated); // Returns true because Value was queried
    Thread.Sleep(10000); // Wait 10 seconds and display the time again
    Console.WriteLine(s.Value); // The delegate is NOT invoked now; same result
}

當我運行這段代碼後，結果以下：

False
2:40:42 PM
True
2:40:42 PM ß Notice that the time did not change 10 seconds later

上述代碼構造Lazy 類的實例，並向它傳遞某個 LazyThreadSafetyMode 標誌。

public enum LazyThreadSafetyMode {
    None, // 徹底沒有線程安全支持，適合GUI 應用程序
    ExecutionAndPublication // Uses the double-check locking technique
    PublicationOnly, // Uses the Interlocked.CompareExchange technique
}

 

內存有限時可能不想建立Lazy 類的實例。這時可調用 System.Threading.LazyInitializer 類的靜態方法。下面展現了這個類：

public static class LazyInitializer {
    // These two methods use Interlocked.CompareExchange internally:
    public static T EnsureInitialized<T>(ref T target) where T: class;
    public static T EnsureInitialized<T>(ref T target, Func<T> valueFactory) where T: class;
    // These two methods pass the syncLock to Monitor's Enter and Exit methods internally
    public static T EnsureInitialized<T>(ref T target, ref Boolean initialized, ref Object syncLock);
    public static T EnsureInitialized<T>(ref T target, ref Boolean initialized,ref Object syncLock, Func<T> valueFactory);
}

另外，爲EnsureInitialized 方法的 syncLock 參數顯式指定同步對象，能夠用同一個鎖保護多個初始化函數和字段。下面展現瞭如何使用這個類的方法：

public static void Main() {
    String name = null;
    // Because name is null, the delegate runs and initializes name
    LazyInitializer.EnsureInitialized(ref name, () => "Jeffrey");
    Console.WriteLine(name); // Displays "Jeffrey"
    
    // Because name is not null, the delegate does not run; name doesn’t change
    LazyInitializer.EnsureInitialized(ref name, () => "Richter");
    Console.WriteLine(name); // Also displays "Jeffrey"
}

 

條件變量模式

假定一個線程但願在一個複合條件爲true 時執行一些代碼。一個選項是讓線程連續「自旋」，反覆測試條件，但這會浪費 CPU時間，也不可能對構成複合條件的多個變量進行原子性的測試。

幸虧有這樣一個模式 容許 線程根據一個複合條件來同步它們的操做，並且不會浪費資源。這個模式稱爲 條件變量模式。

internal sealed class ConditionVariablePattern {
    private readonly Object m_lock = new Object();
    private Boolean m_condition = false;
    public void Thread1() {
        Monitor.Enter(m_lock); // Acquire a mutual-exclusive lock
        // While under the lock, test the complex condition "atomically"
        while (!m_condition) {
        // If condition is not met, wait for another thread to change the condition
        Monitor.Wait(m_lock); // Temporarily release lock so other threads can get it
    }
        // The condition was met, process the data...
        Monitor.Exit(m_lock); // Permanently release lock
   }
    public void Thread2() {
        Monitor.Enter(m_lock); // Acquire a mutual-exclusive lock
        // Process data and modify the condition...
        m_condition = true;
        // Monitor.Pulse(m_lock); // Wakes one waiter AFTER lock is released
        Monitor.PulseAll(m_lock); // Wakes all waiters AFTER lock is released
        Monitor.Exit(m_lock); // Release lock
   }
}

下面展現了一個線程安全的隊列，它容許多個線程在其中對數據項 進行入隊和出對操做。注意，除了有一個可供處理的數據項，不然試圖出隊一個數據項會一直阻塞。

internal sealed class SynchronizedQueue<T> {
    private readonly Object m_lock = new Object();
    private readonly Queue<T> m_queue = new Queue<T>();
    public void Enqueue(T item) {
        Monitor.Enter(m_lock);
        // After enqueuing an item, wake up any/all waiters
        m_queue.Enqueue(item);
        Monitor.PulseAll(m_lock);
        Monitor.Exit(m_lock);
   }
    
    public T Dequeue() {
        Monitor.Enter(m_lock);
        // Loop while the queue is empty (the condition)
        while (m_queue.Count == 0)
            Monitor.Wait(m_lock);
            // Dequeue an item from the queue and return it for processing
        T item = m_queue.Dequeue();
        Monitor.Exit(m_lock);
        return item;
   }
}

 

異步的同步構造

假定客戶端向網站發出請求。客戶端請求到達時，一個線程池線程開始處理客戶端請求。假定這個客戶端想以線程安全的方式修改數據，因此它請求一個 reader-writer 鎖來進行寫入。假定這個鎖被長時間佔有。在鎖佔有期間，另外一個客戶端請求到達了，因此線程池爲這個請求建立新線程。而後，線程阻塞，嘗試獲取 reader-writer 鎖來進行讀取。事實上，隨着愈來愈多的客戶端請求到達，線程池線程會建立愈來愈多的線程，因此這些線程都要傻傻地在鎖上阻塞。服務器把它的全部時間都花在建立線程上面，而目的僅僅是讓它們中止運行！這樣的服務器徹底沒有伸縮性可言。

更糟的是，當Writer 線程釋放鎖時，全部reader線程都同時解除阻塞開始執行。如今又變成了大量線程試圖在相對數量不多的CPU 上運行。因此 ， Windows 開始在線程之間不停地進行上下文切換。因爲上下文切換產生了大量開銷，因此真正的工做反而沒有獲得很好的處理。 這些構造想要解決的許多問題其實最好就是用 第 27 章討論的Task 類來完成。

拿 Barrier 類來講：能夠生成幾個 Task 對象來處理一個階段。而後，當全部這些任務完成後，能夠用另一個或多個 Task 對象繼續。和本章展現的大量構造相比，任務具備下述許多優點。

• 任務使用的內存比線程少得多，建立和銷燬所需的時間也少得多。

• 線程池根據可用CPU 數量自動伸縮任務規模。

• 每一個任務完成一個階段後，運行任務的線程回到線程池，在那裏能接受新任務。

• 線程池是站在整個進程的高度觀察任務，全部，它能更好地調度這些任務，減小進程中的線程數，並減小上下文切換。

 

重點來了：若是代碼能經過異步的同步構造指出它想要一個鎖，那麼會很是有用。在這種狀況下，若是線程得不到鎖，可直接返回並執行其餘工做，而沒必要在那裏傻傻地阻塞。之後當鎖可用時，代碼可恢復執行並訪問鎖保護的資源。

SemaphoreSlim 類經過 WaitAsync 方法實現了這個思路，下面是該方法的最複雜重載版本的簽名。

public Task<Boolean> WaitAsync(Int32 millisecondsTimeout, CancellationToken cancellationToken);

可用它異步地同步對一個資源的訪問（不阻塞任何線程）。

private static async Task AccessResourceViaAsyncSynchronization(SemaphoreSlim asyncLock) {
    // TODO: Execute whatever code you want here...
    
    await asyncLock.WaitAsync(); // Request exclusive access to a resource via its lock
    // When we get here, we know that no other thread is accessing the resource
    // TODO: Access the resource (exclusively)...
    
    // When done accessing resource, relinquish lock so other code can access the resource
    asyncLock.Release();
    // TODO: Execute whatever code you want here...
}

通常建立最大技術爲1 的 SemaphoreSlim，從而對 SemaphoreSlim 保護的資源進行互斥訪問。因此這和使用 Monitor 時的行爲類似，只是 SemaphoreSlim 不支持全部權和遞歸語義。

 

對於 reader-writer 語義， .Net Framework 提供了：ConcurrentExclusiveSchedulerPair 類。

public class ConcurrentExclusiveSchedulerPair {
    public ConcurrentExclusiveSchedulerPair();
    public TaskScheduler ExclusiveScheduler { get; }
    public TaskScheduler ConcurrentScheduler { get; }
    // Other methods not shown...
}

這個類的兩個 TaskScheduler 對象，它們在調度任務時負責 提供 reader/writer 語義。只要當前沒有運行使用 ConcurrentScheduler 調度的任務，使用 ExclusiveScheduler 調度的任何任務將獨佔式地運行。另外，只要當前沒有運行使用 ExclusiveScheduler 調度的任務，使用 ConcurrentScheduler 調度的任務就可同時運行。

private static void ConcurrentExclusiveSchedulerDemo() {
    var cesp = new ConcurrentExclusiveSchedulerPair();
    var tfExclusive = new TaskFactory(cesp.ExclusiveScheduler);
    var tfConcurrent = new TaskFactory(cesp.ConcurrentScheduler);
    
    for (Int32 operation = 0; operation < 5; operation++) {
        var exclusive = operation < 2; // For demo, I make 2 exclusive & 3 concurrent
        
       (exclusive ? tfExclusive : tfConcurrent).StartNew(() => {
            Console.WriteLine("{0} access", exclusive ? "exclusive" : "concurrent");
            // TODO: Do exclusive write or concurrent read computation here...
       });
   }
}

遺憾的是 .Net Framework 沒有提供具備 reader/writer 語義的異步鎖。但做者構建了一個這樣的類， AsyncOneManyLock。用法和SemaphoreSlim 同樣。

 

做者的AsyncOneManyLock 類 內部沒有使用任何內核構造。只使用了一個SpinLock，它在內部使用了用戶模式的構造。 WaitAsync 和 Realse 方法 用鎖保護的只是一些整數計算和比較，以及構造一個 TaskCompletionSource ，並把它添加/刪除 從隊列中。這花不了多少時間，能保證鎖只是短期被佔有。

private static async Task AccessResourceViaAsyncSynchronization(AsyncOneManyLock asyncLock) {
    // TODO: Execute whatever code you want here...
    
    // Pass OneManyMode.Exclusive or OneManyMode.Shared for wanted concurrent access
    await asyncLock.AcquireAsync(OneManyMode.Shared); // Request shared access
    // When we get here, no threads are writing to the resource; other threads may be reading
    // TODO: Read from the resource...
    
    // When done accessing resource, relinquish lock so other code can access the resource
    asyncLock.Release();
    // TODO: Execute whatever code you want here...
}

 

併發集合類

FCL 自帶4個 線程安全的集合類，在 System.Collections.Concurrent 命名空間中定義。它們是：ConcurrentQueue，ConcurrentStack，ConcurrentDictionary 和 ConcurrentBag。

全部這些集合都是「非阻塞」的。換言之，若是一個線程試圖提取一個不存在的元素（數據項），線程會當即返回；線程不會阻塞在那裏，等着一個元素的出現。

一個集合「非阻塞」，並不意味着它就不須要鎖了。ConcurrentDictionary 類在內部使用了 Monitor。ConcurrentQueue 和 ConcurrentStack 確實不須要鎖；它們在內部都使用 Interlocked 的方法來操縱集合。一個 ConcurrentBag 對象由大量迷你集合對象構成，每一個線程一個。

ConcurrentStack，ConcurrentQueue 和 ConcurrentBag 這三個併發集合類都實現了 IProducerConsumerCollection 接口。實現了這個接口的任何類 都能轉變成一個阻塞集合。要將非阻塞的集合轉變爲阻塞集合，須要構造一個System.Collections.Concurrent.BlockingColllection 類，向它的構造器傳遞對非阻塞集合的引用。