C#线程安全集合ConcurrentDictionary

这个在System.Collections.Concurrent的字典类跟字典Dictionary的使用差不多是一样的，但在多线程并发访问的时候，只能靠ConcurrentDictionary了。

这是我之前储存多个SineTones的频率范围使用Dictionary用多线程并发访问：

public static Dictionary<double, Tuple<double, double>> FreqRange { get; set; }

public static void CalcSineTones()
        {

            FreqRange = new Dictionary<double, Tuple<double, double>>();
            Parallel.ForEach(OrderTrackDicResult, pair =>
            {
                SineTonesDic[pair.Key] = CalcAmp(pair.Key, pair.Value);
                SineTonesDicResult[pair.Key] = SineTonesDic[pair.Key];
            });
        }

 private static void CalcAmp(double order, double[][] otrkBuffer)
        {
            double minfreq,maxfreq;
            GetMaxMinFreq(order, otrkBuffer[0], out minfreq, out maxfreq);
            FreqRange[order] = new Tuple<double, double>(minfreq, maxfreq);

        }

结果报System.NullRefrenceException错误：

将Dictionary替换为ConcurrentDictionary：

public static ConcurrentDictionary<double, Tuple<double, double>> FreqRange { get; set; }

public static void CalcSineTones()
        {

            FreqRange = new ConcurrentDictionary<double, Tuple<double, double>>();
            Parallel.ForEach(OrderTrackDicResult, pair =>
            {
                SineTonesDic[pair.Key] = CalcAmp(pair.Key, pair.Value);
                SineTonesDicResult[pair.Key] = SineTonesDic[pair.Key];
            });
        }

 private static void CalcAmp(double order, double[][] otrkBuffer)
        {
            double minfreq,maxfreq;
            GetMaxMinFreq(order, otrkBuffer[0], out minfreq, out maxfreq);
            FreqRange[order] = new Tuple<double, double>(minfreq, maxfreq);

        }

运行没有问题！

ConcurrentBag   集合

表示对象的线程安全的无序集合。

static void Main(string[] args)
{
    ConcurrentBag<int> thList = new ConcurrentBag<int>();
    Parallel.For(0, 100000, a =>
      {
          thList.Add(a);
      });
    thList.TryPeek(out int result1);   //获取末尾的值
    Console.WriteLine(string.Format("Count:{0} Result:{1}", thList.Count, result1));

    thList.TryTake(out int result2);  //删除并获取末尾的值
    Console.WriteLine(string.Format("Count:{0} Result:{1}", thList.Count, result2));
    Console.ReadKey();
}

注：若是使用List<int>,在并行添加数据时要么会抛异常要么集合中的个数不对。

PS：ConcurrentBag<T>不能像List<T>一样轻易获取任何索引处的值和删除任意一个值。 若要使用获取和删除，请使用ConcurrentDictionary。

ConcurrentDictionary 字典

static void Main(string[] args)
{
    ConcurrentDictionary<int, int> thDict = new ConcurrentDictionary<int, int>();
    Parallel.For(0, 10, a =>
      {
          bool isOK = thDict.TryAdd(a, a);
          if (thDict.ContainsKey(4))
          {
              thDict[4] = 400;
          }
      });
    Console.WriteLine(string.Format("Count:{0} Values:{1}", thDict.Count, string.Join(",", thDict.Values)));

    thDict.TryRemove(3, out int value);
    Console.WriteLine(string.Format("Count:{0} value:{1} Values:{2}", thDict.Count, value, string.Join(",", thDict.Values)));

    Console.ReadKey();
}

 ConcurrentQueue  队

表示线程安全的先进先出 (FIFO) 集合。 

static void Main(string[] args)
{
    ConcurrentQueue<int> thQueue = new ConcurrentQueue<int>();
    Parallel.For(0, 10, a =>
    {
        thQueue.Enqueue(a);
    });
    thQueue.TryPeek(out int result1);   //获取开头处的对象
    thQueue.TryDequeue(out int result2);  //获取开头处的对象并将其移除
    Console.ReadKey();
}

ConcurrentStack  栈

表示线程安全的后进先出 (LIFO) 集合。 

static void Main(string[] args)
{
    ConcurrentStack<int> thQueue = new ConcurrentStack<int>();
    Parallel.For(0, 10, a =>
    {
        thQueue.Push(a);
    });
    thQueue.TryPeek(out int result1);   //获取顶部的对象
    thQueue.TryPop(out int result2);  //获取顶部的对象并将其移除
    Console.ReadKey();
}

线程安全的集合所在的命名空间 using System.Collections.Concurrent;

Concurrent意思是并发的，并行的。反义是sequential（顺序的）,线程安全的意思就是多线程中的同步

System.Collections.Concurrent 命名空间

System.Collections.Concurrent 命名空间提供多个线程安全集合类。当有多个线程并发访问集合时，应使用这些类代替 System.Collections 和 System.Collections.Generic 命名空间中的对应类型。 但是，不保证通过扩展方法或通过显式接口实现访问集合对象是线程安全的，可能需要由调用方进行同步。

类

接口

枚举

参考微软官方文档：

https://docs.microsoft.com/zh-cn/dotnet/api/system.collections.concurrent?view=netframework-4.5

参考源代码（类BlockingCollection<T>）

https://referencesource.microsoft.com/#System/sys/system/collections/concurrent/BlockingCollection.cs,0c84922eea8bf511

一、示例：BlockingCollection<T> ，该类与队列ConcurrentQueue<T>类似。

源程序：

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace ThreadSafeCollectionDemo
{
    class Program
    {
        static void Main(string[] args)
        {
            Console.WriteLine("测试阻塞集合BlockingCollection的添加【Add】和移除【Take】...");
            Task task = AddTakeBlockingCollectionAsync();
            Task.WaitAll(task);
            Console.WriteLine("下面测试阻塞集合BlockingCollection的尝试移除【TryTake】...");
            TryTakeBlockingCollection();
            Console.ReadLine();
        }

        /// <summary>
        /// 阻塞集合BlockingCollection的Add(添加)和Take(移除)：生产者【添加元素】、消费者【移除元素】线程安全同时操作
        /// </summary>
        /// <returns></returns>
        static async Task AddTakeBlockingCollectionAsync()
        {
            using (BlockingCollection<int> blockingCollection = new BlockingCollection<int>())
            {
                //生产者
                Task taskProducer = Task.Run(() => 
                {
                    //添加一个元素
                    blockingCollection.Add(1);
                    blockingCollection.Add(2);
                    blockingCollection.Add(3);
                    //结束添加元素：集合已标记为完成添加。CompleteAdding()方法会导致Take()方法出现移除操作异常
                    blockingCollection.CompleteAdding();
                });
                //消费者
                Task taskConsumer = Task.Factory.StartNew(() => 
                {
                    try
                    {
                        while (true)
                        {
                            //移除一个元素，并返回这个移除的元素
                            int removeItem = blockingCollection.Take();
                            Console.WriteLine(removeItem);
                        }
                    }
                    catch (Exception ex)
                    {
                        Console.WriteLine($"移除项出现异常：{ex.Message}");
                    }
                });
                await Task.WhenAll(taskProducer, taskConsumer);
            }
        }

        /// <summary>
        /// 尝试移除阻塞集合BlockingCollection中的元素
        /// </summary>
        static void TryTakeBlockingCollection()
        {
            using (BlockingCollection<int> bc = new BlockingCollection<int>())
            {
                for (int i = 0; i < 1000; i++)
                {
                    bc.Add(i);
                }
                Console.WriteLine($"是否完成添加：【{bc.IsAddingCompleted}】");
                bc.CompleteAdding();//标记集合已经完成添加
                Console.WriteLine($"是否完成添加：【{bc.IsAddingCompleted}】");

                int outerSum = 0;
                Action action = new Action(() =>
                {
                    int localItem;
                    int sum = 0;
                    while (bc.TryTake(out localItem))
                    {
                        sum += localItem;
                    }
                    Interlocked.Add(ref outerSum, sum);
                    Console.WriteLine($"求和Sum={sum}，当前替换：【{outerSum}】");
                });

                //并行操作：三个委托同时执行
                //相当于 将数组求和 分 三小部分数组 分别求和，最后合并
                Parallel.Invoke(action, action, action);
                Console.WriteLine($"此时outerSum为【{outerSum}】");
                Console.WriteLine($"此时集合已完成添加并且集合为空：【{bc.IsCompleted}】");
            }
        }

    }
}

程序运行如图：