在上一章的并行编程实现里，为了保护资源，我们对共享资源加锁（各种同步原语）来进行保护，避免多线程同时访问（主要是写入）。但一般来说，共享资源是一个可以由多个线程读写的集合，即便多线程也应该能够同时写入。因此，使用同步原语对于这种数据集合来说，就不是很合适。

本章将学习线程安全（Thread-Safe）的集合。本章的内容还是比较简单，主要是代码示例较多。

本教程对应学习工程：魔术师Dix / HandsOnParallelProgramming · GitCode

1、并发集合详解

从 .NET Framework 4 开始，许多线程安全集合被添加到 .NET 库中，例如：

IProducerConsumerCollection<T>
BlockingCollection<T>
ConcurrentDictionary<TKey, TValue>

这些都包含在了命名空间 System.Collections.Concurrent 之中。

System.Collections.Concurrent 命名空间 | Microsoft Learn提供多个线程安全集合类，只要多个线程同时访问集合，就应使用这些类来代替 System.Collections 和 System.Collections.Generic 命名空间中的相应类型。但是，不保证通过扩展方法或通过显式接口实现访问集合对象是线程安全的，可能需要由调用方进行同步。 https://learn.microsoft.com/zh-cn/dotnet/api/system.collections.concurrent?view=netstandard-2.1

当使用这些结构时，不需要任何同步，并且读取和更新都将以原子方式完成。并发集合（Concurrent Collection）包装轻量级的 Slim 同步原语，在内核上的负担更轻。

1.1、IProducerConsumerCollection<T>

顾名思义 IProducerConsumerCollection 就是生产者和消费者的集合，为通用同类对提供有效的无锁替代集合。通过继承接口以实现功能：

IProducerConsumerCollection定义供制造者/使用者用来操作线程安全集合的方法。此接口提供一个统一的表示（为生产者/消费者集合），从而更高级别抽象如 BlockingCollection<T> 可以使用集合作为基础的存储机制。 https://learn.microsoft.com/zh-cn/dotnet/api/system.collections.concurrent.iproducerconsumercollection-1?view=netstandard-2.1 首先，我这里先写一个测试用例：

        public static void AddRangeItem(this IProducerConsumerCollection<CaseItem> cases){for (int i = 0; i < 10; i++){var item = new CaseItem();item.Index = i;item.Value = $"{cases.GetType().Name}_{i}";cases.TryAdd(item);}}public static void ParallelDebugCases(this IProducerConsumerCollection<CaseItem> cases){int length = cases.Count;Parallel.For(0, length, x =>{CaseItem item;if (cases.TryTake(out item))Debug.Log($"[{item.Index}] : {item.Value}");});Debug.Log("执行打印完成");}

这里逻辑就很简单，依次添加10个CaseItem，然后并行取出打印出来。我这里直接写一个类，以不做任何并行处理的方式实现，然后直接运行查看结果：

可以看到，打印的结果就比较诡异了，没有打印出 1 和 7 。也就是有 2 个 CaseItem 在并行过程中丢失了。如果要想在多线程中正常使用，显然是要我们自己提供线程同步的方案。

如果我们使用传统的队列（Queue），那么就需要像这样写类似的代码：

        public bool TryTake(out CaseItem item){lock (m_LockObject){if (m_Queue.Count > 0){item = m_Queue.Dequeue();return true;}item = null;return false;}}

我这里是直接加锁了。当然也可以使用 Monitor 的写法，或者其他同步原语。但是我们发现这样确实不方便，毕竟重复的代码很多。

那么下面作者介绍了几个内置的类可供使用：

1.2、并发队列 ConcurrentQueue<T>

并发队列 (ConcurrentQueue<T>) 是队列（Queue<T>）的线程安全版本，而不必自己写同步原语。

ConcurrentQueue表示线程安全的先进先出 (FIFO) 集合。 https://learn.microsoft.com/zh-cn/dotnet/api/system.collections.concurrent.concurrentqueue-1?view=netstandard-2.1 而且，ConcurrentQueue<T> 是已经继承了 IProducerConsumerCollection<T>：

using System.Collections.Generic;namespace System.Collections.Concurrent
{public class ConcurrentQueue<T> : IProducerConsumerCollection<T>, IEnumerable<T>, IEnumerable, ICollection, IReadOnlyCollection<T>{public ConcurrentQueue();public ConcurrentQueue(IEnumerable<T> collection);public int Count { get; }public bool IsEmpty { get; }public void Clear();public void CopyTo(T[] array, int index);public void Enqueue(T item);public IEnumerator<T> GetEnumerator();public T[] ToArray();public bool TryDequeue(out T result);public bool TryPeek(out T result);}
}

在以下应用场景中，考虑使用并发队列：

每个项目的处理时间都很短。
只有一个专用生产者线程或只有一个专用消费者线程。
处理速度在 500 FLOPS 或以上。

上面是书上的说法，确实比较晦涩。这里提供一个参考文章：

C#线程 ConcurrentQueue安全队列介绍_Paddy Pan的博客-CSDN博客c#安全线程队列ConcurrentQueus介绍及方法实现原理和使用 _concurrentqueuehttps://blog.csdn.net/qq_41230604/article/details/126305068

这个就讲得比较清楚了。例如在添加项目进队列时，不是直接Lock而是使用了一次自旋锁。总的来说，只要是高并发需求，使用 ConcurrentQueue<T> 是更优的选择。但是如果没有并发，ConcurrentQueue<T> 的性能应该是不如 Queue<T> 的。

之前做过一个性能测试，可以看下具体差异：

【学习积累】Queue 与 ConcurrentQueue性能测试_concurrentqueue和queue的区别_魔术师Dix的博客-CSDN博客在 C# 中，关于队列（Queue）有两种，一种就是我们普通使用的队列，另一种是线程安全的队列ConcurrentQueue 。本文将对这两个队列进行一个简单的性能测试，同时讨论一种特殊情况：一个线程入队，一个线程出队时使用 Queue 的情况。_concurrentqueue和queue的区别https://blog.csdn.net/cyf649669121/article/details/130162599

这里写一个测试用例：

        private void RunWithConcurrentQueue(){ConcurrentQueue<CaseItem> caseItems = new ConcurrentQueue<CaseItem>();caseItems.AddRangeItem();caseItems.ParallelDebugCases();}

执行结果如下：

可见10次打印刚好就是 0~9 ，没有任何重复项。

1.3、并发堆栈 ConcurrentStack<T>

ConcurrentStack<T> 是 Stack<T> 的并发版本，同样也继承了 IProducerConsumerCollection 接口。当然，与队列的区别，就是堆栈是 LIFO（后进先出），而队列是FIFO（先进先出）。同样的，并发堆栈不涉及内存锁定，靠自旋（Spinning）和比较交换（Compare-And_Swap，CAS）无锁算法来消除竞争。

ConcurrentStack表示线程安全的后进先出 (LIFO) 集合。 https://learn.microsoft.com/zh-cn/dotnet/api/system.collections.concurrent.concurrentstack-1?view=netstandard-2.1 这里我们用 1.2 的代码改造一下进行示例：

     private void RunWithConcurrentStack(){ConcurrentStack<CaseItem> caseItems = new ConcurrentStack<CaseItem>();caseItems.AddRangeItem();caseItems.ParallelDebugCases();}

运行结果也是相同的：

1.4、ConcurrentBag<T>

这个和前两个类似，但是是无序集合。

ConcurrentBag表示对象的线程安全的无序集合。 https://learn.microsoft.com/zh-cn/dotnet/api/system.collections.concurrent.concurrentbag-1?view=netstandard-2.1

ConcurrentBag<T> 已针对同一线程既充当生产者又充当消费者的情况进行了优化。而且还支持工作窃取（详见任务并行性：11、工作窃取队列）算法，为每一个线程维护一个本地队列。

似乎比较抽象，这里写一段代码测试一下：

        private void RunWtihConcurrentBag(){ConcurrentBag<int> caseItems = new ConcurrentBag<int>();ManualResetEventSlim resetEventSlim = new ManualResetEventSlim(false);Task t1 = Task.Run(() =>{for (int i = 0; i < 5; i++){caseItems.Add(i);}resetEventSlim.Wait();int length = caseItems.Count;for (int i = 0; i < length; i++){int val;caseItems.TryTake(out val);Debug.Log($"case item : {val}");}});Task t2 = Task.Run(() =>{for (int i = 5; i < 10; i++){caseItems.Add(i);}resetEventSlim.Set();});}

运行结果如下所示：

可见在 Bag 内部明显分了 2 个部分，一个部分是 0~4，一部分是 5~9，也就是分别由 t1 和 t2 添加的事项。0~4 的事项总是排列在一起，但相互之间是无序的；5~9 的事项也是如此。按照书上说法，ConcurrentBag<T>为每一个线程维护了一个本地队列，都是优先完成自己线程的队列，再处理其他线程的，所以打印结果才会如上图所示。

1.5、BlockingCollection<T>

BlockingCollection<T> 可以限制生产者线程生成的最大数目，然后生产者线程将睡眠，并进入阻塞。当消费者线程使用了项目时，生产者线程将接触阻塞；当集合被耗尽时，消费者线程将被阻塞。

BlockingCollection为实现 IProducerConsumerCollection<T> 的线程安全集合提供阻塞和限制功能。 https://learn.microsoft.com/zh-cn/dotnet/api/system.collections.concurrent.blockingcollection-1?view=netstandard-2.1

综上 BlockingCollection<T> 有两个概念：

限制：集合达到最大值时，不能添加任何新对象，生产者线程进入睡眠模式。
阻塞：当集合为空时，可以阻塞消费者线程。
这里写个代码来进行示例：

        //创建  BlockingCollection 并设定容量为 5，填0代表无上限private BlockingCollection<CaseItem> m_BlockingCaseItems = new BlockingCollection<CaseItem>(5);// 生产者线程private void RunWithBlockingCollectionProduce(){Task.Run(() =>{for (int i = 0; i < 10; i++){CaseItem item = new CaseItem();item.Index = i;item.Value = $"[{i}]_BlockCollection";m_BlockingCaseItems.Add(item);//如果当前容量 大于5 ，则会在这里阻塞；Debug.Log($"添加一个事项：{item.Value} / {m_BlockingCaseItems.Count}");}Debug.Log("全部添加完成 ！ ");});}//消费者线程private void RunWtithBlockingCollectionConsumer(){Task.Run(async () =>{for (int i = 0; i < 5; i++){CaseItem item = m_BlockingCaseItems.Take();//如果当前容量为0，则会在在这里阻塞Debug.LogWarning($"消费一个事项：{item.Value}");await Task.Delay(1000);}Debug.LogWarning("全部消费完成 ！ ");});}

这两个方法我们有两个顺序调用，我们先调用生产者线程，再调用消费者线程，效果如下：

可以看到，在添加了5个事项之后，达到容量上限之后不再继续添加，线程直接阻塞。当消费者线程每取出一个事项，生产者线程则解除一次阻塞并继续执行。

现在我们反过来，先调用消费者线程，再调用生产者线程，结果如下：

当开始调用消费者线程时，由于容量为空，所以直接阻塞了消费者线程。开始生产者线程之后，便开始唤醒并进行消费，从上图的 Log 中可以清楚地看到两个线程的相互唤醒作用。

2、多生产者-消费者应用场景

这一节其实说的就是 BlockingCollection<T> 的两个语法糖：

namespace System.Collections.Concurrent
{public class BlockingCollection<T>:IEnumerable<T>, IEnumerable, IReadOnlyCollection<T>, ICollection, IDisposable{public static int AddToAny(BlockingCollection<T>[] collections, T item);public static int TakeFromAny(BlockingCollection<T>[] collections, out T item);}
}

也就是可以从一个集合中获取、添加事项。

3、ConcurrentDictionary<TKey,TValue>

很显然了，ConcurrentDictionary<TKey,TValue> 就是用于多线程并行的字典，可以消除键的重复问题。这个字典读是无锁的，而写入是上锁的。

ConcurrentDictionary表示可由多个线程同时访问的键/值对的线程安全集合。 https://learn.microsoft.com/zh-cn/dotnet/api/system.collections.concurrent.concurrentdictionary-2?view=netstandard-2.1 这里写个代码示例：

        private void RunWithConcurrentDictionary(){ConcurrentDictionary<int, CaseItem> dictCaseItems = new ConcurrentDictionary<int, CaseItem>();var t1 = Task.Run(() =>{Parallel.For(0, 5, i =>{CaseItem item = new CaseItem();item.Index = i;item.Value = $"T1 : {i}";dictCaseItems.AddOrUpdate(i, item, (k, v) =>{Debug.Log($"T1 冲突: {v.Value} |");return v;});});});var t2 = Task.Run(() =>{Parallel.For(4, 8, i =>{CaseItem item = new CaseItem();item.Index = i;item.Value = $"T2 : {i}";dictCaseItems.AddOrUpdate(i, item, (k, v) =>{Debug.Log($"T2 冲突: {v.Value} |");return v;});});});var t3 = Task.Run(() =>{Parallel.For(6, 10, i =>{CaseItem item = new CaseItem();item.Index = i;item.Value = $"T3 : {i}";dictCaseItems.AddOrUpdate(i, item, (k, v) =>{Debug.Log($"T3 冲突: {v.Value} |");return v;});});});Task.WaitAll(t1, t2, t3);Debug.Log("最后结果");foreach (var item in dictCaseItems.Values){Debug.Log(item.Value);}}

执行打印结果如下：