编程需要对基本的数据结构和算法有所了解。程序员为并发情况选择最合适的数据结 构，那就需要知道很多事情，例如算法运行时间、空间复杂度，以及大写0标记法等。在不 同的广为人知的场景中，我们总知道哪种数据结构更高效。对于并行计算，我们需要使用适当的数据结构。这些数据结构具备可伸缩性，尽可能地 避免锁，同时还能提供线程安全的访问。.NET framework版本4引入了 System.Collections. Concunent命名空间，其中包含了一些数据结构。在本章中，我们将展示这些数据结构并通 过简单的例子来说明如何使用它们
a. ConcurrentQueue
ConcurrentQueue集合使用了原子的比较和交换（Compare and Swap, 简称CAS）操作，以及SpinWait来保证线程安全。它实现了一个先进先出（First In First Out,简称FIFO)的集合，这意味着元素出队列的顺序与加入队列的顺序是一致的。可以 调用Enqueue方法向队列中加入元素 TryDequeue方法试图取出队列中的第一个元素，而 TryPeek方法则试图得到第一个元素但并不从队列中删除该元素
using System; using System.Collections.Concurrent; using System.Threading; using System.Threading.Tasks; using static System.Console;namespace Chapter6.Recipe2 {class Program{static void Main(string[] args){Task t = RunProgram();t.Wait();}static async Task RunProgram(){//使用ConcurrentQueue集合实例创建了一个任务队列var taskQueue = new ConcurrentQueue<CustomTask>();//创建了 一个取消标志，它是用来在我们将任务放入队列后停止工作的var cts = new CancellationTokenSource();//接下来启动了一个单独的工作者线程来将任务放入任务队列中var taskSource = Task.Run(() => TaskProducer(taskQueue));//现在定义该程序中消费任务的部分。我们创建了四个工作者、它们会随机等待一段时 间，然后从任务队列中获取一个任务，处理该任务，//一直重复整个过程直到我们发出取消标 志信号Task[] processors = new Task[4];for (int i = 1; i <= 4; i++){string processorId = i.ToString();processors[i-1] = Task.Run(() => TaskProcessor(taskQueue, $"Processor {processorId}", cts.Token));}await taskSource;cts.CancelAfter(TimeSpan.FromSeconds(2));//我们看到队列中的任务按从前到后的顺序被处理，但一个后面的任务是有可能会比前面 的任务先处理的，因为我们有四个工作者独立地运行，//而且任务处理时间并不是恒定的。我 们看到访问该队列是线程安全的，没有一个元素会被提取两次await Task.WhenAll(processors);Read();}static async Task TaskProducer(ConcurrentQueue<CustomTask> queue){for (int i = 1; i <= 20; i++){await Task.Delay(50);var workItem = new CustomTask {Id = i};queue.Enqueue(workItem);WriteLine($"Task {workItem.Id} has been posted");}}static async Task TaskProcessor(ConcurrentQueue<CustomTask> queue, string name, CancellationToken token){CustomTask workItem;bool dequeueSuccesful = false;await GetRandomDelay();do{dequeueSuccesful = queue.TryDequeue(out workItem);if (dequeueSuccesful){WriteLine($"Task {workItem.Id} has been processed by {name}");}await GetRandomDelay();}while (!token.IsCancellationRequested);}static Task GetRandomDelay(){int delay = new Random(DateTime.Now.Millisecond).Next(1, 500);return Task.Delay(delay);}class CustomTask{public int Id { get; set; }}} }
b. Concurrentstack
Concurrentstack的实现也没有使用任何锁，只釆用了 CAS操作。它是一个后进先出 (Last In First Out,简称LIFO)的集合，这意味着最近添加的元素会先返回。可以使用Push 和PushRange方法添加元素，使用TryPop和TryPopRange方法获取元素，以及使用TryPeek 方法检查元素。
c. ConcurrentBag
ConcurrentBag是一个支持重复元素的无序集合。它针对这样以下情况进行了优化.即 多个线程以这样的方式工作：每个线程产生和消费自己的任务，极少与其他线程的任务交互 (如果要交互则使用锁)。添加元素使用Add方法，检查元素使用TryPeek方法、获取元素使 用TryTake方法，请避免使用上面提及的集合的Count属性。实现这些集合使用的是链表，Count操作 的时间复杂度为0(N)。如果想检查集合是否为空，请使用IsEmpty属性，其时间复 杂度为0(1)。
d. ConcurrentDictionary
ConcurrentDictionary是一个线程安全的字典集合的实现。对于读操作无需使用锁。但是 对于写操作则需要锁，该并发字典使用多个锁，在字典桶之上实现了一个细粒度的锁模型。 使用参数concurrencyLevel可以在构造函数中定义锁的数量、这意味着预估的线程数量将并 发地更新该字典。就由于并发字典使用锁，所以一些操作需要获取该字典中的所有锁：如果没必要请避免 使用以下操作：Count, IsEmpty. Keys、Values. CopyTo 及 To Array。
using System.Collections.Concurrent; using System.Collections.Generic; using System.Diagnostics; using static System.Console;namespace Chapter6.Recipe1 {class Program{const string Item = "Dictionary item";const int Iterations = 1000000;public static string CurrentItem;static void Main(string[] args){//当程序启动时我们创建了两个集合，其中一个是标准的字典集合，另一个是新的并发字典集合var concurrentDictionary = new ConcurrentDictionary<int, string>();var dictionary = new Dictionary<int, string>();var sw = new Stopwatch();sw.Start();for (int i = 0; i < Iterations; i++){//采用锁的机制向标准的字典中添加元素，并测量完成100万次迭代的时间lock (dictionary){dictionary[i] = Item;}}sw.Stop();WriteLine($"Writing to dictionary with a lock: {sw.Elapsed}");//Writing to dictionary with a lock: 00:00:00.0449525 sw.Restart();for (int i = 0; i < Iterations; i++){//采用同样的场景来测量ConcurrentDictionary的性能，concurrentDictionary[i] = Item;}sw.Stop();WriteLine($"Writing to a concurrent dictionary: {sw.Elapsed}");//Writing to a concurrent dictionary: 00:00:00.2662710//最后比较从两个集合中获取值的性能 sw.Restart();for (int i = 0; i < Iterations; i++){lock (dictionary){CurrentItem = dictionary[i];}}sw.Stop();WriteLine($"Reading from dictionary with a lock: {sw.Elapsed}");//Reading from dictionary with a lock: 00:00:00.0192289 sw.Restart();for (int i = 0; i < Iterations; i++){CurrentItem = concurrentDictionary[i];}sw.Stop();WriteLine($"Reading from a concurrent dictionary: {sw.Elapsed}");//Reading from a concurrent dictionary: 00:00:00.0108079 ReadLine();}//通过这个非常简单的场景，我们发现ConcurrentDictionary写操作比使用锁的通常 的字典要慢得多，//而读操作则要快些。因此如果对字典需要大量的线程安全的读操作, ConcurrentDictionary 是最好的选择//如果你对字典只需要多线程访问只读元素，则没必要执行线程安全的读操作。在此场 景中最好只使用通常的字典或ReadOnlyDictionary集合。//ConcurrentDictionary的实现使用了细粒度锁(fine-grained locking)技术，这在多线程写 入方面比使用锁的通常的字典(也被称为粗粒度锁)的可伸缩性更好。//正如本例中所示，当 只用一个线程时，并发字典非常慢，但是扩展到5到6个线程(如果有足够的CPU核心来同 时运行它们)，并发字典的性能会更好 } }
e. BlockingCollection
BlockingCollection 是对 IProducerConsumerCollection 泛型接 口的实现的一个高级封装。 它有很多先进的功能来实现管道场景，即当你有一些步骤需要使用之前步骤运行的结果时: BlockingCollectione类支持如下功能：分块、调整内部集合容量、取消集合操作、从多个块 集合中获取元素
