.Net多线程编程—并发集合

并发集合

1 为什么使用并发集合?

原因主要有以下几点:

  • System.Collections和System.Collections.Generic名称空间中所提供的经典列表、集合和数组都不是线程安全的,若无同步机制,他们不适合于接受并发的指令来添加和删除元素。
  • 在并发代码中使用上述经典集合需要复杂的同步管理,使用起来很不方便。
  • 使用复杂的同步机制会大大降低性能。
  • NET Framework 4所提供的新的集合尽可能地减少需要使用锁的次数。这些新的集合通过使用比较并交换(compare-and-swap,CAS)指令和内存屏障,避免使用互斥的重量级锁。这对性能有保障。

注意:

与经典集合相比,并发集合会有更大的开销,因此在串行代码中使用并发集合无意义,只会增加额外的开销且运行速度比访问经典集合慢。

2 并发集合

1)ConcurrentQueue:线程安全的先进先出 (FIFO) 集合

主要方法:

  • Enqueue(T item);将对象添加到集合结尾。
  • TryDequeue(out T result); 尝试移除并返回位于集合开始处的对象,返回值表示操作是否成功。
  • TryPeek(out T result);尝试返回集合开始处的对象,但不将其移除,返回值表示操作是否成功。

说明:

  • ConcurrentQueue是完全无锁的,但当CAS操作失败且面临资源争用时,它可能会自旋并且重试操作。
  • ConcurrentQueue是FIFO集合,某些和出入顺序无关的场合,尽量不要用ConcurrentQueue。

2)ConcurrentStack:线程安全的后进先出 (LIFO) 集合

主要方法及属性:

  • Push(T item);将对象插入集合的顶部。
  • TryPop(out T result);尝试弹出并返回集合顶部的对象,返回值表示操作是否成功。
  • TryPeek(out T result);尝试返回集合开始处的对象,但不将其移除,返回值表示操作是否成功。
  • IsEmpty { get; }指示集合是否为空。
  • PushRange(T[] items);将多个对象插入集合的顶部。
  • TryPopRange(T[] items);弹出顶部多个元素,返回结果为弹出元素个数。

说明:

  • 与ConcurrentQueue相似地,ConcurrentStack完全无锁的,但当CAS操作失败且面临资源争用时,它可能会自旋并且重试操作。
  • 获取集合是否包含元素使用IsEmpty属性,而不是通过判断Count属性是否大于零。调用Count比调用IsEmpty开销大。
  • 使用PushRange(T[] items)和TryPopRange(T[] items)时注意缓冲引起的额外开销和额外的内存消耗。

3) ConcurrentBag:元素可重复的无序集合

主要方法及属性:

  • TryPeek(out T result);尝试从集合返回一个对象,但不移除该对象,返回值表示是否成功获得该对象。
  • TryTake(out T result);尝试从集合返回一个对象并移除该对象,返回值表示是否成功获得该对象。
  • Add(T item);将对象添加到集合中。
  • IsEmpty { get; }解释同ConcurrentStack

说明:

  • ConcurrentBag为每一个访问集合的线程维护了一个本地队列,在可能的情况下,它会以无锁的方式访问本地队列。
  • ConcurrentBag在同一个线程添加和删除元素的场合下效率非常高。
  • 因为ConcurrentBag有时会需要锁,在生产者线程和消费者线程完全分开的场景下效率非常低。
  • ConcurrentBag调用IsEmpty的开销非常大,因为这需要临时获得这个无序组的所有锁。

4)BlockingCollection:实现

System.Collections.Concurrent.IProducerConsumerCollection<T> 的线程安全集合,提供阻塞和限制功能

主要方法及属性:

  • BlockingCollection(int boundedCapacity);boundedCapacity表示集合限制大小。
  • CompleteAdding();将BlockingCollection实例标记为不再接受任何添加。
  • IsCompleted { get; }此集合是否已标记为已完成添加并且为空。
  • GetConsumingEnumerable();从集合中移除并返回移除的元素
  • Add(T item);添加元素到集合。
  • TryTake(T item, int millisecondsTimeout, CancellationToken cancellationToken);

说明:

  • 使用BlockingCollection()构造函数实例化BlockingCollection,意味着不设置boundedCapacity,那么boundedCapacity为默认值: int.MaxValue。
  • 限界:使用BlockingCollection(int boundedCapacity),设置boundedCapacity的值,当集合容量达到这个值得时候,向BlockingCollection添加元素的线程将会被阻塞,直到有元素被删除。

限界功能可控制内存中集合最大大小,这对于需要处理大量元素的时候非常有用。

  • 默认情况下,BlockingCollection封装了一个ConcurrentQueue。可以在构造函数中指定一个实现了IProducerConsumerCollection接口的并发集合,包括:ConcurrentStack、ConcurrentBag。
  • 使用此集合包含易于无限制等待的风险,所以使用TryTake更加,因为TryTake提供了超时控制,指定的时间内可以从集合中移除某个项,则为 true;否则为 false。

5)ConcurrentDictionary:可由多个线程同时访问的键值对的线程安全集合。

主要方法

  • AddOrUpdate(TKey key, TValue addValue, Func<TKey, TValue, TValue> updateValueFactory);如果指定的键尚不存在,则将键/值对添加到 字典中;如果指定的键已存在,则更新字典中的键/值对。
  • GetOrAdd(TKey key, TValue value);如果指定的键尚不存在,则将键/值对添加到字典中。
  • TryRemove(TKey key, out TValue value);尝试从字典中移除并返回具有指定键的值。
  • TryUpdate(TKey key, TValue newValue, TValue comparisonValue);将指定键的现有值与指定值进行比较,如果相等,则用第三个值更新该键。

说明:

  • ConcurrentDictionary对于读操作是完全无锁的。当多个任务或线程向其中添加元素或修改数据的时候,ConcurrentDictionary使用细粒度的锁。使用细粒度的锁只会锁定真正需要锁定的部分,而不是整个字典。

6)IProducerConsumerCollection:定义供生产者/消费者用来操作线程安全集合的方法。 此接口提供一个统一的表示(为生产者/消费者集合),从而更高级别抽象如 System.Collections.Concurrent.BlockingCollection<T>可以使用集合作为基础的存储机制。

3.常用模式

1)并行的生产者-消费者模式

定义:

生成者和消费者是此模式中的两类对象模型,消费者依赖于生产者的结果,生产者生成结果的同时,消费者使用结果。

图1 并行的生产者-消费者模式

说明:

  • 并发集合用在此模式下非常合适,因为并发集合支持此模式中对象的并行操作。
  • 若不使用并发集合,那么就要加入同步机制,从而使程序变得比较复杂,难于维护和理解,同时大大降低性能。
  • 上图为生产者消费者模式示意图,纵轴为时间轴,生成者与消费者的并不在一条时间线上,但二者有交叉,意在表明生成者先产生结果,而后消费者才真正使用了生成者产生的数据。

2)流水线模式

定义:

流水线由多个阶段构成,每个阶段由一系列的生产者和消费者构成。一般来讲前一个阶段是后一个阶段的生成者;依靠相邻两个阶段之间的缓冲区队列,每个阶段可以并发执行。

图2 并行的流水线模式

说明:

  • 常使用BlockingCollection<T>作为缓冲罐区队列。
  • 流水线的速度近似等于流水线最慢阶段的速度。
  • 上图为流水线模式示意图,前一阶段为后一阶段的生成者,这里展示了最为简单和基本的流水线模式,更复杂的模式可以认为是每个阶段都包括了对数据更多的处理过程。

4 使用方式

仅以ConcurrentBag和BlockingCollection为例,其他的并发集合与之相似。

ConcurrentBag

1 List<string> list = ......
2 ConcurrentBag<string> bags = new ConcurrentBag<string>();
3 Parallel.ForEach(list, (item) => 
4 {
5     //对list中的每个元素进行处理然后,加入bags中
6     bags.Add(itemAfter);
7 });

BlockingCollection—生产者消费者模式

 1 public static void Execute()
 2 {
 3             //调用Invoke,使得生产者任务和消费者任务并行执行
 4             //Producer方法和Customer方法在Invoke中的参数顺序任意,不论何种顺序都会获得正确的结果
 5             Parallel.Invoke(()=>Customer(),()=>Producer());
 6             Console.WriteLine(string.Join(",",customerColl));
 7 }
 8 
 9 //生产者集合
10 private static BlockingCollection<int> producerColl = new BlockingCollection<int>();
11  //消费者集合
12 private static BlockingCollection<string> customerColl = new BlockingCollection<string>();
13 
14 public static void Producer()
15 {
16             //循环将数据加入生成者集合
17             for (int i = 0; i < 100; i++)
18             {
19                 producerColl.Add(i);
20             }
21 
22             //设置信号,表明不在向生产者集合中加入新数据
23             //可以设置更加复杂的通知形式,比如数据量达到一定值且其中的数据满足某一条件时就设置完成添加
24             producerColl.CompleteAdding();
25 }
26 
27 public static void Customer()
28 {
29             //调用IsCompleted方法,判断生产者集合是否在添加数据,是否还有未"消费"的数据
30             //注意不要使用IsAddingCompleted,IsAddingCompleted只表明集合标记为已完成添加,而不能说明其为空
31             //而IsCompleted为ture时,那么IsAddingCompleted为ture且集合为空
32             while (!producerColl.IsCompleted)
33             {
34                 //调用Take或TryTake "消费"数据,消费一个,移除一个
35                 //TryAdd的好处是提供超时机制
36                 customerColl.Add(string.Format("消费:{0}", producerColl.Take()));
37             }
38 }

-----------------------------------------------------------------------------------------

时间仓促,水平有限,如有不当之处,欢迎指正。

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏令仔很忙

OutOfMemoryError异常----Java堆溢出

在Java虚拟机规范的描述中,除了程序计数器外,虚拟机内存的其他几个运行时区域都有发生OutOfMemoryError(下面都叫OOM)异常的肯能,下面就通过...

772
来自专栏博客园

X--名称空间详解

X名称空间里面的成员(如X:Name,X:Class)都是写给XAML编译器看的、用来引导XAML代码将XAML代码编译为CLR代码。

922
来自专栏大内老A

深入理解C# 3.x的新特性(2):Extension Method[上篇]

在C#3.0中,引入了一些列新的特性,比如: Implicitly typed local variable, Extension method,Lambda ...

1816
来自专栏鹅厂少年的奇妙之旅

Go内存模型

Go语言中内存分配大致有3种模式:Stack、Heap、Fixed Size Segment。

2734
来自专栏木木玲

堆外内存 之 DirectByteBuffer 详解

2097
来自专栏Java成长之路

二、OutOfMemoryError实战

本文通过一些可执行代码来验证异常发生的场景,并且会初步介绍几个与内存相关的最基本的虚拟机参数。 本文的主要目的有两个: 1. 通过代码验证Java虚拟机...

522
来自专栏大内老A

ASP.NET MVC的Model元数据提供机制的实现

在前面的介绍中我们已经提到过表示Model元数据的ModelMetadata对象最终是通过一个名为ModelMetadataProvider的组件提供的,接下来...

1816
来自专栏程序员的SOD蜜

条件表达式的短路求值与函数的延迟求值

延迟求值是 .NET的一个很重要的特性,在LISP语言,这个特性是依靠宏来完成的,在C,C++,可以通过函数指针来完成,而在.NET,它是靠委托来完成的。如果不...

1936
来自专栏CSDN技术头条

知名互联网公司 Java 开发岗面试知识点解析

互联网公司的校园招聘,从形式上说,面试一般分为 2-3 轮技术面试 +1 轮 HR 面试。

1012
来自专栏冰霜之地

神经病院Objective-C Runtime住院第二天—消息发送与转发

现在越来越多的app都使用了JSPatch实现app热修复,而JSPatch 能做到通过 JS 调用和改写 OC 方法最根本的原因是 Objective-C 是...

1113

扫码关注云+社区