第C#中的并發(fā)集合Concurrent類System.Collections.Concurrent命名空間提供多個線程安全集合類。

當有多個線程并發(fā)訪問集合時，應(yīng)使用這些類代替System.Collections和System.Collections.Generic命名空間中的對應(yīng)類型。

為了對集合進行線程安全的訪問，定義了IProducerConsumerCollection接口。這個接口中最重要的方法是TryAdd()和TryTake()。

TryAdd()方法嘗試給集合添加一項，但如果集合禁止添加項，這個操作就可能失敗。為了給出相關(guān)信息，TryAdd()方法返回一個布爾值，以說明操作是成功還是失敗。TryTake()方法也以這種方式工作，以通知調(diào)用者操作是成功還是失敗，并在操作成功時返回集合中的項。

二、空間中包含的類

ConcurrentXXX：這些集合是線程安全的，如果某個動作不適用于線程的當前狀態(tài)，它們就返回false。在繼續(xù)之前，總是霈要確認添加或提取元素是否成功。不能相信集合會完成任務(wù)。

1、ConcurrentQueue隊列

這個集合類用一種免鎖定的算法實現(xiàn)，使用在內(nèi)部合并到一個鏈表中的32項數(shù)組。

訪問隊列元素的方法有Enqueue()、TryDequeue()和TryPeek()。這些方法的命名非常類似于前面Queue類的方法，只是給可能調(diào)用失敗的方法加上了前綴Try。

因為這個類實現(xiàn)了IProducerConsumerCollection接口，所以TryAdd()和TryTake()方法僅調(diào)用Enqueue()和TryDequeue()方法。

2、ConcurrentStack堆棧

非常類似于ConcurrentQueue類，只是帶有另外的元素訪問方法。

ConcurrcntStack類定義了Push()、PushRange()、TryPeek()、TiyPop()和TryPopRange()方法。在內(nèi)部這個類使用其元素的鏈表。

3、ConcurrentBag包

該類沒有定義添加或提取項的任何順序。這個類使用一個把線程映射到內(nèi)部使用的數(shù)組上的概念，因此嘗試減少鎖定。

訪問元素的方法有Add()、TryPeek()和TryTake()。

4、ConcurrentDictionary字典

這是一個線程安全的鍵值集合。

TryAdd()、TryGetValue()、TryRemove()和TryUpdate()方法以非阻塞的方式訪問成員。

因為元素基于鍵和值,所以ConcurrentDictionary沒有實現(xiàn)IProducerConsumerCollection。

5、BlockingCollection集合

這個集合在可以添加或提取元素之前，會阻塞線程并一直等待。

BlockingCollection集合提供了一個接口,以使用Add()和Take()方法來添加和刪除元素。這些方法會阻寒線程，一直等到任務(wù)可以執(zhí)行為止。

Add()方法有一個重載版本，其中可以給該重載版本傳遞一個CancellationToken令牌。

這個令牌允許取消被阻塞的調(diào)用。如果不希望線程無限期地等待下去，且不希望從外部取消調(diào)用，就可以使用TryAdd()和TryTake()方法，在這些方法中，也可以指定一個超時值，它表示在調(diào)用失敗之前應(yīng)阻塞線程和等待的最長時間。

6、BlockingCollection阻塞式集合

這是對實現(xiàn)了IProducerConsumerCollection接口的任意類的修飾器，它默認使用ConcurrentQueue類。

還可以給構(gòu)造函數(shù)傳遞任何其他實現(xiàn)了IProducerConsumerCollection接口的類。

下面的代碼示例簡單演示了使用BlockingCollection類和多個線程的過程。

一個線程是生成器，它使用Add()方法給集合寫入元素，另一個線程是使用者，它使用Take()方法從集合中提取元素：

internalstaticBlockingCollectionint_TestBCollection;

classThreadWork1//生產(chǎn)者

publicThreadWork1()

publicvoidrun()

System.Console.WriteLine("ThreadWork1run{");

for(inti=0;i100;i++)

System.Console.WriteLine("ThreadWork1producer:"+i);

_TestBCollection.Add(i);

_TestBCollection.CompleteAdding();

System.Console.WriteLine("ThreadWork1run}");

classThreadWork2//消費者

publicThreadWork2()

publicvoidrun()

inti=0;

intnCnt=0;

boolIsDequeuue=false;

System.Console.WriteLine("ThreadWork2run{");

while(!_TestBCollection.IsCompleted)

IsDequeuue=_TestBCollection.TryTake(outi);

if(IsDequeuue)

System.Console.WriteLine("ThreadWork2consumer:"+i*i+"====="+i);

nCnt++;

System.Console.WriteLine("ThreadWork2run}");

staticvoidStartT1()

ThreadWork1work1=newThreadWork1();

work1.run();

staticvoidStartT2()

ThreadWork2work2=newThreadWork2();

work2.run();

staticvoidMain(string[]args)

Taskt1=newTask(()=StartT1());

Taskt2=newTask(()=StartT2());

_TestBCollection=newBlockingCollectionint//可以跟容量

Console.WriteLine("Sample4-4Main{");

Console.WriteLine("Maint1t2started{");

t1.Start();

t2.Start();

Console.WriteLine("Maint1t2started}");

Console.WriteLine("Mainwaitt1t2end{");

Task.WaitA