首页 > > 网络编程 > 其它 >

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

2018-06-17 22:44:01来源：未知阅读 ()

本随笔续接：.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> 提供了几种方法中添加或更新键/值对在字典中下, 如表中所述。

任务	使用此方法	用法说明
如果它尚不在字典中存在向字典中，添加新的密钥	TryAdd	如果当前不在字典中存在该键，此方法将添加指定的键/值对。该方法返回 true 或 false具体取决于是否已添加新对。
如果该注册表项具有特定值，更新为现有键在字典中值	TryUpdate	此方法检查是否密钥具有指定的值，如果它存在，则用新值更新该键。它相当于CompareExchange 方法，但它的用于字典的元素。
无条件地将键/值对存储在字典中，覆盖已存在的键的值	索引器的资源库︰dictionary[key] = newValue
将键/值对添加到字典中，或如果键已存在，更新基于键的现有值的键的值	AddOrUpdate(TKey, Func<TKey, TValue>, Func<TKey, TValue, TValue>) - 或 - AddOrUpdate(TKey, TValue, Func<TKey, TValue, TValue>)	AddOrUpdate(TKey, Func<TKey, TValue>, Func<TKey, TValue, TValue>) 接受的键和两个委托。如果键在字典; 中不存在，则使用第一个委托它接受键并返回应添加的键的值。如果该键不存在; 它使用第二个委托它接受的键和其当前值，并返回应为项设置的新值。 AddOrUpdate(TKey, TValue, Func<TKey, TValue, TValue>) 接受键、值要添加，以及更新委托。这是与以前的重载中，相同之处在于它不使用委托来添加的键。
获取此键在字典中，向字典中添加值并将其返回如果该键不存在的值	GetOrAdd(TKey, TValue) - 或 - GetOrAdd(TKey, Func<TKey, TValue>)	这些重载提供延迟初始化为键/值对在字典中，添加的值，仅当不存在。 GetOrAdd(TKey, TValue) 采用键不存在要添加的值。 GetOrAdd(TKey, Func<TKey, TValue>) 将一个委托，可将生成的值，如果键不存在。