作為多線程和并行計算不得不考慮的問題就是臨界資源的訪問問題，解決臨界資源的訪問通常是加鎖或者是使用信號量，這個大家應(yīng)該很熟悉了。

　　而集合作為一種重要的臨界資源，通用性更廣，為了讓大家更安全的使用它們，微軟為我們帶來了強大的并行集合：System.Collections.Concurrent里面的各位仁兄們。

　　首先，咱們從一個經(jīng)典的問題談起。

生產(chǎn)者消費者問題

　　這個問題是最為經(jīng)典的多線程應(yīng)用問題，簡單的表述這個問題就是：有一個或多個線程(生產(chǎn)者線程)產(chǎn)生一些數(shù)據(jù)，同時，還有一個或者多個線程(消費者線程)要取出這些數(shù)據(jù)并執(zhí)行一些相應(yīng)的工作。如下圖所示：

　　下面就是使用程序去描述這個問題了。

　　最直接的想法可能是這樣：

static void Main(string[] args) {int count = 0;// 臨界資源區(qū)var queue = new Queue<string>();// 生產(chǎn)者線程Task.Factory.StartNew(() =>{while (true){queue.Enqueue("value" + count);count++;}});// 消費者線程1Task.Factory.StartNew(() =>{while (true){if (queue.Count > 0){string value = queue.Dequeue();Console.WriteLine("Worker 1: " + value);}}});// 消費者線程2Task.Factory.StartNew(() =>{while (true){if (queue.Count > 0){string value = queue.Dequeue();Console.WriteLine("Worker 2: " + value);}}});Thread.Sleep(50000); }

　　使用Queue<string>模擬了一個簡單的資源池，一個生產(chǎn)者放數(shù)據(jù)，兩個消費者消費數(shù)據(jù)。上面這個程序運行以后會產(chǎn)生異常，異常的原因很簡單，當(dāng)某個時刻，第一個消費者判斷queue.Count > 0為true時，就會到Queue中取數(shù)據(jù)，但是這個時候數(shù)據(jù)可能會被第二個消費者拿走了，因為第二個消費者也判斷出此時有數(shù)據(jù)可取。這是一個簡單的臨界資源線程安全問題。

　　知道問題了，那么如何解決呢？
　　第一種方案是加鎖，這個方案是可行的，很多時候我們也是這么做的，包括微軟早期實現(xiàn)線程安全的ArrayList和Hashtable內(nèi)部(Synchronized方法)也是這么實現(xiàn)的。這個方案適用于只有少量的消費者，并且每個消費者都會執(zhí)行大量操作的時候，這時lock并沒什么太大問題，但是，如果是大批量短小精悍的消費者存在的話，lock會嚴(yán)重影響代碼的執(zhí)行效率。
　　第二種方案就是我們直接用新的線程安全的集合區(qū)解決這個問題。新的線程安全的這些集合內(nèi)部不再使用lock機制這種比較低效的方式去實現(xiàn)線程安全，而是轉(zhuǎn)而使用SpinWait和Interlocked等機制，間接實現(xiàn)了線程安全，這種方式的效率要高于使用lock的方式。看一下實現(xiàn)代碼：

var queue = new ConcurrentQueue<string>(); Task.Factory.StartNew(() => {while (true){queue.Enqueue("value" + count);count++;} });Task.Factory.StartNew(() => {while (true){string value;if (queue.TryDequeue(out value)){Console.WriteLine("Worker 1: " + value);}} });Task.Factory.StartNew(() => {while (true){string value;if (queue.TryDequeue(out value)){Console.WriteLine("Worker 2: " + value);}} });

　　執(zhí)行這段代碼，可以工作，但是有點不太優(yōu)雅，能不能不要去判斷集合是否為空？集合當(dāng)自己沒有元素的時候自己Block一下可以嗎？答案當(dāng)然是可以的，使用BlockingCollection即可：

var blockingCollection = new BlockingCollection<string>(); Task.Factory.StartNew(() => {while (true){blockingCollection.Add("value" + count);count++;} });Task.Factory.StartNew(() => {while (true){Console.WriteLine("Worker 1: " + blockingCollection.Take());} });Task.Factory.StartNew(() => {while (true){Console.WriteLine("Worker 2: " + blockingCollection.Take());} });

　　BlockingCollection集合是一個擁有阻塞功能的集合，它就是完成了經(jīng)典生產(chǎn)者消費者的算法功能。它沒有實現(xiàn)底層的存儲結(jié)構(gòu)，而是使用了實現(xiàn)IProducerConsumerCollection接口的幾個集合作為底層的數(shù)據(jù)結(jié)構(gòu)，例如ConcurrentBag， ConcurrentStack或者是ConcurrentQueue。你可以在構(gòu)造BlockingCollection實例的時候傳入這個參數(shù)，如果不指定的話，則默認(rèn)使用ConcurrentQueue作為存儲結(jié)構(gòu)。

　　而對于生產(chǎn)者來說，只需要通過調(diào)用其Add方法放數(shù)據(jù)，消費者只需要調(diào)用Take方法來取數(shù)據(jù)就可以了。
　　當(dāng)然了上面的消費者代碼中還有一點是讓人不爽的，那就是while語句，可以更優(yōu)雅一點嗎？答案還是肯定的：

Task.Factory.StartNew(() => {foreach (string value in blockingCollection.GetConsumingEnumerable()){Console.WriteLine("Worker 1: " + value);} });

　　GetConsumingEnumerable()方法是關(guān)鍵，這個方法會遍歷集合取出數(shù)據(jù)，一旦發(fā)現(xiàn)集合空了，則阻塞自己，直到集合中又有元素了再開始遍歷，神奇吧。

　　好了，到此完美了解決了生產(chǎn)者消費者問題。然而通常來說，還有兩個問題我們有時需要去控制：

第一個問題：控制集合中數(shù)據(jù)的最大數(shù)量。

　　這個問題由BlockingCollection構(gòu)造函數(shù)解決，構(gòu)造該對象實例的時候，構(gòu)造函數(shù)中的BoundedCapacity決定了集合最大的可容納數(shù)據(jù)數(shù)量，這個比較簡單，不多說了。

第二個問題：何時停止的問題。

　　這個問題由CompleteAdding和IsCompleted兩個配合解決。
　　CompleteAdding方法是直接不允許任何元素被加入集合；當(dāng)使用了CompleteAdding方法后且集合內(nèi)沒有元素的時候，另一個屬性IsCompleted此時會為True，這個屬性可以用來判斷是否當(dāng)前集合內(nèi)的所有元素都被處理完。看一下生產(chǎn)者修改后的代碼：

Task.Factory.StartNew(() => {for (int count = 0; count < 10; count++){blockingCollection.Add("value" + count);}blockingCollection.CompleteAdding(); });

　　當(dāng)使用了CompleteAdding方法后，對象停止往集合中添加數(shù)據(jù)，這時如果是使用GetConsumingEnumerable枚舉的，那么這種枚舉會自然結(jié)束，不會再Block住集合，這種方式最優(yōu)雅，也是推薦的寫法。但是如果是使用TryTake訪問元素的，則需要使用IsCompleted判斷一下，因為這個時候使用TryTake會拋InvalidOperationException異常。

看一下最終的代碼形式：

static void Main(string[] args) {var blockingCollection = new BlockingCollection<string>();var producer = Task.Factory.StartNew(() =>{for (int count = 0; count < 10; count++){blockingCollection.Add("value" + count);Thread.Sleep(300);}blockingCollection.CompleteAdding();});var consumer1 = Task.Factory.StartNew(() =>{foreach (string value in blockingCollection.GetConsumingEnumerable()){Console.WriteLine("Worker 1: " + value);}});var consumer2 = Task.Factory.StartNew(() =>{foreach (string value in blockingCollection.GetConsumingEnumerable()){Console.WriteLine("Worker 2: " + value);}});Task.WaitAll(producer, consumer1, consumer2); }

BlockingCollection的枚舉

　　此外，需要注意BlockingCollection有兩種枚舉方法，首先BlockingCollection本身繼承自IEnumerable<T>，所以它自己就可以被foreach枚舉，首先BlockingCollection包裝了一個線程安全集合，那么它自己也是線程安全的，而當(dāng)多個線程在同時修改或訪問線程安全容器時，BlockingCollection自己作為IEnumerable會返回一個一定時間內(nèi)的集合片段，也就是只會枚舉在那個時間點上內(nèi)部集合的元素。使用這種方式枚舉的時候，不會有Block效果。
　　另外一種方式就是我們上面使用的GetConsumingEnumerable方式的枚舉，這種方式會有Block效果，直到CompleteAdding被調(diào)用為止。

　　最后提一下實現(xiàn)IProducerConsumerCollection接口的幾個集合：ConcurrentBag(線程安全的無序的元素集合)， ConcurrentStack(線程安全的堆棧)和ConcurrentQueue(線程安全的隊列)。這些都很簡單，功能與非線程安全的那些集合都一樣，只不多是多了TryXXX方法，多線程環(huán)境下使用這些方法就好了，其他就不多說了。

　　到此生產(chǎn)者和消費者這個經(jīng)典的問題告一段落了。

　　System.Collections.Concurrent下面的集合除了解決生產(chǎn)者消費者問題外，還有一些與多線程相關(guān)的集合，例如：

1.?ConcurrentDictionary，這個是鍵/值對字典的線程安全實現(xiàn)，這個類在原來的基礎(chǔ)上也添加了一下新的方法，例如：AddOrUpdate，GetOrAdd，TryXXX等等，都很容易理解。

2. 各種Partitioner 類，提供針對數(shù)組、列表和可枚舉項的常見分區(qū)策略。

　　若要對數(shù)據(jù)源操作進(jìn)行并行化，其中一個必要步驟是將源分區(qū)為可由多個線程同時訪問的多個部分。 PLINQ 和任務(wù)并行庫 (TPL) 提供了默認(rèn)的分區(qū)程序，當(dāng)編寫并行查詢或ForEach循環(huán)時，默認(rèn)的分區(qū)程序以透明方式工作。 但是毫無疑問，對于一些復(fù)雜的情況，我們是可以插入自己的分區(qū)程序的，這就是微軟為我們提供的各種Partitioner類，這個不多說了，感興趣的同學(xué)請自己參考一下MSDN。

?

總結(jié)

以上是生活随笔為你收集整理的C#的变迁史07 - C# 4.0 之线程安全集合篇的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。