.NET 同步与异步 之 线程安全的集合 (十一)

本随笔续接：.NET 同步与异步 之 警惕闭包（十）

无论之前说的锁、原子操作 还是 警惕闭包，都是为安全保驾护航，本篇随笔继续安全方面的主题：线程安全的集合。

先看一下命名空间：System.Collections.Concurrent，常用的类型有（均为泛型）：BlockingCollection<T>、ConcurrentBag<T>、ConcurrentDictionary<TKey, TValue>、ConcurrentQueue<T>、ConcurrentStack<T> 。

其中：ConcurrentBag<T> 为无序的集合、ConcurrentDictionary<TKey, TValue> 为词典类型。

ConcurrentQueue<T>、ConcurrentStack<T>  分别为队列 和 栈，而 BlockingCollection<T> 可以看做是 队列 和 栈的进一步封装调用，并提供了阻塞（超时）功能。

本随笔着重说两个类型：BlockingCollection<T>  和 ConcurrentDictionary<TKey, TValue>。

一、BlockingCollection<T>

1、先看一下 MSDN 上的Demo

        /// <summary>
        /// MSDN Demo
        /// BlockingCollection<T>.Add()
        /// BlockingCollection<T>.CompleteAdding()
        /// BlockingCollection<T>.TryTake()
        /// BlockingCollection<T>.IsCompleted
        /// </summary>
        public void Demo1()
        {
            // Construct and fill our BlockingCollection
            using (BlockingCollection<int> bc = new BlockingCollection<int>())
            {
                int NUMITEMS = 10000;

                for (int i = 0; i < NUMITEMS; i++)
                {
                    bc.Add(i);
                }
                bc.CompleteAdding();

                int outerSum = 0;

                // Delegate for consuming the BlockingCollection and adding up all items
                Action action = () =>
                {
                    int localItem;
                    int localSum = 0;

                    while (bc.TryTake(out localItem))
                    {
                        localSum += localItem;
                    }
                    Interlocked.Add(ref outerSum, localSum);
                };

                // Launch three parallel actions to consume the BlockingCollection
                Parallel.Invoke(action, action, action);

                base.PrintInfo(string.Format("Sum[0..{0}) = {1}, should be {2}", NUMITEMS, outerSum, ((NUMITEMS * (NUMITEMS - 1)) / 2)));
                base.PrintInfo(string.Format("bc.IsCompleted = {0} (should be true)", bc.IsCompleted));

            }
        }

MSDN Demo

从demo中看一下 BlockingCollection<T> 的用法

1）Add 方法， 将项添加到集合中。

2）CompleteAdding 方法，标记当前实例不可以再添加任何项。

3）TryTake 方法，如果可以从当前集合移除一个项，则返回true,否则返回False. 如果该集合为空，则此方法立即返回 false。

　　删除了某个项的顺序取决于用于创建集合的类型 BlockingCollection<T> 实例。 当您创建 BlockingCollection<T> 对象，您可以指定要使用的集合类型(通过构造函数指定)。 例如，可以指定 ConcurrentQueue<T> 先进先出 (FIFO) 行为的对象或 ConcurrentStack<T> 后进先出 (LIFO) 行为的对象。 可以使用任何集合类来实现 IProducerConsumerCollection<T> 接口。 默认集合类型 BlockingCollection<T> 是 ConcurrentQueue<T>。

4）IsCompleted 属性，获取此 BlockingCollection<T> 是否已标记为完成添加（即 调用了 CompleteAdding 方法）并且为空。

2、限制容量

        /// <summary>
        /// 限制容量
        /// </summary>
        public void Demo2()
        {
            BlockingCollection<int> blocking = new BlockingCollection<int>(5);

            Task.Run(() =>
            {
                for (int i = 0; i < 20; i++)
                {
                    blocking.Add(i);
                    PrintInfo($"add:({i})");
                }

                blocking.CompleteAdding();
                PrintInfo("CompleteAdding");
            });

            // 等待先生产数据
            Task.Delay(500).ContinueWith((t) =>
            {
                while (!blocking.IsCompleted)
                {
                    var n = 0;
                    if (blocking.TryTake(out n))
                    {
                        PrintInfo($"TryTake:({n})");
                    }
                }

                PrintInfo("IsCompleted = true");
            });

        }

限制容量

调研Add方法的时候，如果集合中的项的数量已经达到上限，则Add方法将会被阻塞。

3、在 BlockingCollection  中使用Stack

         /// <summary>
        /// 在 BlockingCollection  中使用Stack
        /// </summary>
        public void Demo3()
        {
            BlockingCollection<int> blocking = new BlockingCollection<int>(new ConcurrentStack<int>(), 5);

            Task.Run(() =>
            {
                for (int i = 0; i < 20; i++)
                {
                    blocking.Add(i);
                    PrintInfo($"add:({i})");
                }

                blocking.CompleteAdding();
                PrintInfo("CompleteAdding");
            });

            // 等待先生产数据
            Task.Delay(500).ContinueWith((t) =>
            {
                while (!blocking.IsCompleted)
                {
                    var n = 0;
                    if (blocking.TryTake(out n))
                    {
                        PrintInfo($"TryTake:({n})");
                    }
                }

                PrintInfo("IsCompleted = true");
            });

        }

在 BlockingCollection 中使用Stack

该Demo和之前的Demo的唯一区别就是：构造函数、指定了 ConcurrentStack<T>  类型为存储容器。

除此之外、BlockingCollection<T> 还是提供了对超时的控制，例如：TryAdd(T, TimeSpan) 、 TryTake(T, TimeSpan) 等数个重载版本。

二、ConcurrentDictionary<TKey, TValue>。

ConcurrentDictionary<TKey, TValue> 类， 已经实现 IDictionary<TKey, TValue> 接口。也就是说 它也实现了Dictionary 类型的基础功能。

此外， ConcurrentDictionary<TKey, TValue> 提供了几种方法中添加或更新键/值对在字典中下, 如表中所述。

这些操作是原子性操作，而且都是线程安全的。在 ConcurrentDictionary<TKey, TValue> 类中 唯一的例外是 AddOrUpdate 和 GetOrAdd 方法，它们是使用细粒度锁定，以确保线程安全。

ConcurrentDictionary<TKey, TValue> 类 在上述的MSDN文档中 已经介绍的差不多了，不再举例。 当然还要提一句， 该类型不支持阻塞操作。

三、线程安全警告

在命名空间 System.Collections.Concurrent 中的类型，都遵循如下的线程安全规则：线程安全集合本身提供的方法 是线程安全的，但是通过其类型实现的接口的方法 和 扩展方法 却不是线程安全的。

随笔暂告一段落、下一篇随笔:  因地制宜——CPU密集型操作和IO密集型操作

附，Demo : http://files.cnblogs.com/files/08shiyan/ParallelDemo.zip

参见更多：随笔导读：同步与异步