在上篇文章《.net中的並行編程-1.基礎知識》中列出了在.net進行多核或並行編程中需要的基礎知識,今天就來分析在基礎知識樹中一個比較簡單常用的並發數據結構--.net類庫中無鎖棧的實現。
首先解釋一下什么這里“無鎖”的相關概念。
所謂無鎖其實就是在普通棧的實現方式上使用了原子操作，原子操作的原理就是CPU在系統總線上設置一個信號，當其他線程對同一塊內存進行訪問時CPU監測到該信號存在會，然后當前線程會等待信號釋放后才能對內存進行訪問。原子操作都是由操作系統API實現底層由硬件支持，常用的操作有：原子遞增，原子遞減，比較交換，ConcurrentStack中的實現就是使用了原子操作中的比較交換操作。
使用原子操作的好處：
第一、由於沒有使用鎖，可以避免死鎖。
第二、原子操作不會阻塞線程，例如執行某個指令時當前線程掛起了（或執行了一次上下文切換），其他線程還能繼續操作，如果使用lock鎖，當前線程掛起后由於沒有釋放鎖，其他線程進行操作時會被阻塞。
第三、由於原子操作直接由硬件指令的支持，所以原子操作性能比普通鎖的高。
使用原子操作的壞處：
第一，使用原子操作一般失敗時會使用回退技術對當前操作進行重試，所以容易產生活鎖和線程飢餓問題，但可以通過隨機退讓等技術進行緩解，但不能消除。
第二，程序員開發使用難度較大，測試難度較大。
下面開始進入正題：
由於.net 中的 ConcurrentStack的代碼較多所以本文就不貼出所有代碼，本人也只分析筆者認為重要的幾個部分，全部源碼可以再去以下微軟官方網址查看
http://referencesource.microsoft.com/#mscorlib/system/Collections/Concurrent/ConcurrentStack.cs
傳統的棧結構都一般都使用單鏈表實現（.net中的Stack使用的是數組）， 入棧操作就是把頭節點替換為新節點，出棧操作就是把頭結點指向下一個節點。所以當大量線程並發訪問時線程的競爭條件都在頭結點也就是說如果我們能報保證對於頭結點操作時是安全的那么整個棧就是安全的。
入棧操作 public void Push(T item)
public void Push(T item) { Node newNode = new Node(item); newNode.m_next = m_head; if (Interlocked.CompareExchange(ref m_head, newNode, newNode.m_next) == newNode.m_next) { return; } // If we failed, go to the slow path and loop around until we succeed. PushCore(newNode, newNode); } private void PushCore(Node head, Node tail) { SpinWait spin = new SpinWait(); // Keep trying to CAS the exising head with the new node until we succeed. do { spin.SpinOnce(); // Reread the head and link our new node. tail.m_next = m_head; } while (Interlocked.CompareExchange(ref m_head, head, tail.m_next) != tail.m_next); }
(在原版的注釋中我們就能看到入棧使用了原子操作中的比較交換（CAS）操作)
入棧分為了三步：
a.當數據入棧時會分配一個新結點，然后將此刻當前內存中的頭結點作為新結點的下一個結點， newNode.m_next = m_head中保持的是當前頭結點的快照也就是說另一個線程此時有可能更改了m_head指向的結點，注意頭結點（m_head）的字段聲明中前面使用了volatile關鍵字，我們知道volatile關鍵字有兩個作用：第一個是禁止編譯器和CPU更改字段的位置，第二個是強制刷新CPU的高速緩存，當讀取該聲明該關鍵字的字段時每次都去內存里重新加載數據然后讀到CPU的高速緩存而不使用CPU緩存中較老的數據，這個地方m_head使用volatile是因為當運行在其他核心的CPU線程更改了m_head值,而我們當前核心的CPU高速緩存中沒有及時更新的問題，還有就是出棧時防止對m_head的操作語句移動到其他語句之后造成邏輯代碼沒有按照預先的邏輯走，例如newNode.m_next = m_head操作放到了if語句之后造成邏輯錯誤或者CPU的指令亂序執行時產生的邏輯錯誤。
b.比較當前的頭結點是否與我們保存的newNode.m_next 的快照頭結點相同，如果相同則將新結點替換為頭結點否則比較失敗進行c步驟。
Interlocked.CompareExchange(ref m_head, newNode, newNode.m_next)其實等價於以下代碼，只不過該代碼的執行是以原子的方式執行。
if (m_head == newNode.m_next) { m_head = newNode.m_next;} else { return m_head; }
c.如果b步驟失敗則進入PushCore（Node head, Node tail）；PushCore的步驟其實就是重復執行步驟B中的Interlocked.CompareExchange(ref m_head, newNode, newNode.m_next)操作,直到新節點寫入為止，在循環期間使用了spin.SpinOnce() ，使用這個API其實就是為了防止活鎖，交換失敗的線程會進行退讓，就好比兩個人迎面走來，可能會發生這種情況：你向左走他也向左走，你向右他也向右，所以為了避免倆人碰到一起，那么一個人可以先原地停止，然后繼續再走，當繼續走時發現方向還是相同時可以改變一下停止的時間，比如說先停止5秒，然后如果還是方向相同則停止10秒，然后還是方向相同可以時可以去喝杯茶慢慢再走，其實這種思想就是隨機退讓。當然計算機中的SpinOnce實現沒有想象的那么簡單，在當分析到同步原語SpinWati的實現時我會詳細介紹SpinOnce方法的實現方式，這個地方就理解為讓線程休息一會（當分析SpinOnce時源碼時發現了一個小細節 Thread,Sleep(0)和Thread.Sleep(1)的使用，其實我發現很多人都不清楚這里0和1的區別，這里要說明一下：Thread,Sleep(0)表示交出當前線程的時間片，讓具有相同或者更高優先級的線程運行否則繼續運行當前線程，也就是說如果沒有相同級別或更高級別的線程等待運行那么還是運行當前線程，這個地方會產生線程飢餓問題。Thread,Sleep(1)表示線程睡眠1毫秒和其他線程優先級沒有關系，其實這里設置為1時也不是睡眠1毫秒而是13毫秒或者更長，具體的和系統的時鍾周期有關)。
出棧操作 public bool TryPop(out T result)
出棧操作也是使用的CAS操作,不斷的將頭結點指向下一個結點然后返回頭結點，如果交換失敗則循環進行，循環期間也是使用隨機退讓技術來較少活鎖的概率只不過退讓的時間會隨着退讓的次數而增大。棧操作的API設計為TryPop（）返回值為bool,，是因為在多個線程同時出棧的過程中有可能一個線程出棧以后棧就空了，所以在出棧有可能失敗。
批量入棧操作 public void PushRange(T[] items, int startIndex, int count)
批量入棧是在單個入棧基礎上實現的，將多個項目入棧時先將壓入的多個項目組成一個棧，然后再將頭結點使用CAS操作指向該生成的棧，所以說批量調用一次入棧的效率要比單個調用多次入棧的效率高。 題外話：這里有個編寫代碼的小細節，在批量入棧的方法中會首先調用ValidatePushPopRangeInput（items, startIndex, count）方法來校驗傳入的參數正確性，其實該編碼是編寫代碼時比較重要的原則叫“手術室原則”，其解釋為醫生進入手術室時，對手套，身體等已經進行了消毒，這些准備工作已經完成了，剩下了工作就是醫生專心完成手術。這種編碼原則不僅使代碼的整潔性提高，而且減少CPU分支預判的次數以提升代碼運行速度，所以在日常編碼中我們可以使用該方法，將大量的參數判斷抽象到單獨的方法中。
判斷是否為空IsEmpty屬性
該操作的實現比較簡單，只要判斷頭結點為空即可。在微軟的文檔中我們發現當我們判斷棧中元素是否為空時應該使用該屬性而不是使用Count == 0 這種判斷方式，因為Count統計棧內元素的個數時，每使用一次會遍歷整個棧，時間復雜度為O(N)而IsEmpty為O(1)，所以使用Count==0 效率比較底下尤其是在數據量大的情況下。
IEnumerable<T>接口成員的實現GetEnumerator()
該方法實現是拿到頭結點然后依次遍歷整個棧，注意該方法拿到了只是當前時刻整個棧的快照，在遍歷過程中棧內元素的增加或減少對於GetEnumerator()返回值的數量不會改變。
其他問題
1.在.net編寫無鎖代碼時不用考慮ABA問題，因為這是.net的垃圾回收來保證的，除非使用了對象池技術，例如將內部分配結點的操作由對象池來負責。
2.在.net源碼中普通的Stack<T> 內部使用的是數組實現，而ConcurretStack內部使用的是鏈表，主要原因還是在於Stack 在使用數組擴容時會有拷貝數據的開銷，尤其是在數據量大的情況下這種性能損失還是比較大的，還有個原因是內部使用鏈表可以避免ABA問題（前提是分配內部結點時沒有使用對象池），不過鏈表的實現也不是沒有缺點，例如入棧時我們會分配一個新結點，而該結點出棧完以后會由GC回收掉，這種結點這時候就成為了垃圾結點，不過在實現ConcurretQueue的時候因為隊列的先進后出的特性使用了另一種解決方案--鏈表+數組的方式，這種方式既解決了垃圾結點的問題又解決了數組擴容復制數據產生的性能開銷問題。
最后，在我們閱讀.net源碼的過程中其實可以發現很多非常經典的編碼技巧和編碼風格，讓我們看代碼時可以由上到下如行雲流水般一氣呵成可，這也是我比較推崇的代碼風格--要像寫詩一樣寫自己代碼，讓別人像讀詩一樣讀你的代碼。
時間不早了就到這了，下片文章中我會繼續分析.net中另外一個比較經典的並發數據結構ConcurrentQueue的實現。
由於筆者能力有限，有分析錯誤的地方難免發生，歡迎大家指正。