[译]何时使用 Parallel.ForEach,何时使用 PLINQ
原作者: Pamela Vagata, Parallel Computing Platform Group, Microsoft Corporation
原文pdf:http://download.csdn.NET/detail/sqlchen/7509513
====================================================================
简介
当需要为多核机器进行优化的时候,最好先检查下你的程序是否有处理能够分割开来进行并行处理。(例如,有一个巨大的数据集合,其中的元素需要一个一个进行彼此独立的耗时计算)。
.net framework 4 中提供了 Parallel.ForEach 和 PLINQ 来帮助我们进行并行处理,本文探讨这两者的差别及适用的场景。
Parallel.ForEach
Parallel.ForEach 是 foreach 的多线程实现,他们都能对 IEnumerable<T> 类型对象进行遍历,Parallel.ForEach 的特殊之处在于它使用多线程来执行循环体内的代码段。
Parallel.ForEach 最常用的形式如下:
public static ParallelLoopResult ForEach<TSource>( IEnumerable<TSource> source, Action<TSource> body)
PLINQ
PLINQ 也是一种对数据进行并行处理的编程模型,它通过 LINQ 的语法来实现类似 Parallel.ForEach 的多线程并行处理。
场景一:简单数据 之 独立操作的并行处理(使用 Parallel.ForEach)
示例代码:
public static void IndependentAction(IEnumerable<T> source, Action<T> action) { Parallel.ForEach(source, element => action(element)); }
理由:
1. 虽然 PLINQ 也提供了一个类似的 ForAll 接口,但它对于简单的独立操作太重量化了。
2. 使用 Parallel.ForEach 你还能够设定 ParallelOptions.MaxDegreeOfParalelism 参数(指定最多需要多少个线程),这样当 ThreadPool 资源匮乏(甚至当可用线程数<MaxDegreeOfParalelism)的时候, Parallel.ForEach 依然能够顺利运行,并且当后续有更多可用线程出现时,Parallel.ForEach 也能及时地利用这些线程。PLINQ 只能通过WithDegreeOfParallelism 方法来要求固定的线程数,即:要求了几个就是几个,不会多也不会少。
场景二:顺序数据 之 并行处理(使用 PLINQ 来维持数据顺序)
当输出的数据序列需要保持原始的顺序时采用 PLINQ 的 AsOrdered 方法非常简单高效。
示例代码:
public static void GrayscaleTransformation(IEnumerable<Frame> Movie) { var ProcessedMovie = Movie .AsParallel() .AsOrdered() .Select(frame => ConvertToGrayscale(frame)); foreach (var grayscaleFrame in ProcessedMovie) { // Movie frames will be evaluated lazily } }
理由:
1. Parallel.ForEach 实现起来需要绕一些弯路,首先你需要使用以下的重载在方法:
public static ParallelLoopResult ForEach<TSource >( IEnumerable<TSource> source, Action<TSource, ParallelLoopState, Int64> body)
这个重载的 Action 多包含了 index 参数,这样你在输出的时候就能利用这个值来维持原先的序列顺序。请看下面的例子:
public static double [] PairwiseMultiply(double[] v1, double[] v2) { var length = Math.Min(v1.Length, v2.Lenth); double[] result = new double[length]; Parallel.ForEach(v1, (element, loopstate, elementIndex) => result[elementIndex] = element * v2[elementIndex]); return result; }
你可能已经意识到这里有个明显的问题:我们使用了固定长度的数组。如果传入的是 IEnumerable 那么你有4个解决方案:
(1) 调用 IEnumerable.Count() 来获取数据长度,然后用这个值实例化一个固定长度的数组,然后使用上例的代码。
(2) The second option would be to materialize the original collection before using it; in the event that your input data set is prohibitively large, neither of the first two options will be feasible.(没看懂贴原文)
(3) 第三种方式是采用返回一个哈希集合的方式,这种方式下通常需要至少2倍于传入数据的内存,所以处理大数据时请慎用。
(4) 自己实现排序算法(保证传入数据与传出数据经过排序后次序一致)
2. 相比之下 PLINQ 的 AsOrdered 方法如此简单,而且该方法能处理流式的数据,从而允许传入数据是延迟实现的(lazy materialized)
场景三:流数据 之 并行处理(使用 PLINQ)
PLINQ 能输出流数据,这个特性在一下场合非常有用:
1. 结果集不需要是一个完整的处理完毕的数组,即:任何时间点下内存中仅保持数组中的部分信息
2. 你能够在一个单线程上遍历输出结果(就好像他们已经存在/处理完了)
示例:
public static void AnalyzeStocks(IEnumerable<Stock> Stocks) { var StockRiskPortfolio = Stocks .AsParallel() .AsOrdered() .Select(stock => new { Stock = stock, Risk = ComputeRisk(stock)}) .Where(stockRisk => ExpensiveRiskAnalysis(stockRisk.Risk)); foreach (var stockRisk in StockRiskPortfolio) { SomeStockComputation(stockRisk.Risk); // StockRiskPortfolio will be a stream of results } }
这里使用一个单线程的 foreach 来对 PLINQ 的输出进行后续处理,通常情况下 foreach 不需要等待 PLINQ 处理完所有数据就能开始运作。
PLINQ 也允许指定输出缓存的方式,具体可参照 PLINQ 的 WithMergeOptions 方法,及 ParallelMergeOptions 枚举
场景四:处理两个集合(使用 PLINQ)
PLINQ 的 Zip 方法提供了同时遍历两个集合并进行结合元算的方法,并且它可以与其他查询处理操作结合,实现非常复杂的机能。
示例:
public static IEnumerable<T> Zipping<T>(IEnumerable<T> a, IEnumerable<T> b) { return a .AsParallel() .AsOrdered() .Select(element => ExpensiveComputation(element)) .Zip( b .AsParallel() .AsOrdered() .Select(element => DifferentExpensiveComputation(element)), (a_element, b_element) => Combine(a_element,b_element)); }
示例中的两个数据源能够并行处理,当双方都有一个可用元素时提供给 Zip 进行后续处理(Combine)。
Parallel.ForEach 也能实现类似的 Zip 处理:
public static IEnumerable<T> Zipping<T>(IEnumerable<T> a, IEnumerable<T> b) { var numElements = Math.Min(a.Count(), b.Count()); var result = new T[numElements]; Parallel.ForEach(a, (element, loopstate, index) => { var a_element = ExpensiveComputation(element); var b_element = DifferentExpensiveComputation(b.ElementAt(index)); result[index] = Combine(a_element, b_element); }); return result; }
当然使用 Parallel.ForEach 后你就得自己确认是否要维持原始序列,并且要注意数组越界访问的问题。
场景五:线程局部变量
Parallel.ForEach 提供了一个线程局部变量的重载,定义如下:
public static ParallelLoopResult ForEach<TSource, TLocal>( IEnumerable<TSource> source, Func<TLocal> localInit, Func<TSource, ParallelLoopState, TLocal,TLocal> body, Action<TLocal> localFinally)
使用的示例:
public static List<R> Filtering<T,R>(IEnumerable<T> source) { var results = new List<R>(); using (SemaphoreSlim sem = new SemaphoreSlim(1)) { Parallel.ForEach(source, () => new List<R>(), (element, loopstate, localStorage) => { bool filter = filterFunction(element); if (filter) localStorage.Add(element); return localStorage; }, (finalStorage) => { lock(myLock) { results.AddRange(finalStorage) }; }); } return results; }
线程局部变量有什么优势呢?请看下面的例子(一个网页抓取程序):
public static void UnsafeDownloadUrls () { WebClient webclient = new WebClient(); Parallel.ForEach(urls, (url,loopstate,index) => { webclient.DownloadFile(url, filenames[index] + ".dat"); Console.WriteLine("{0}:{1}", Thread.CurrentThread.ManagedThreadId, url); }); }
通常第一版代码是这么写的,但是运行时会报错“System.NotSupportedException -> WebClient does not support concurrent I/O operations.”。这是因为多个线程无法同时访问同一个 WebClient 对象。所以我们会把 WebClient 对象定义到线程中来:
public static void BAD_DownloadUrls () { Parallel.ForEach(urls, (url,loopstate,index) => { WebClient webclient = new WebClient(); webclient.DownloadFile(url, filenames[index] + ".dat"); Console.WriteLine("{0}:{1}", Thread.CurrentThread.ManagedThreadId, url); }); }
修改之后依然有问题,因为你的机器不是服务器,大量实例化的 WebClient 迅速达到你机器允许的虚拟连接上限数。线程局部变量可以解决这个问题:
public static void downloadUrlsSafe() { Parallel.ForEach(urls, () => new WebClient(), (url, loopstate, index, webclient) => { webclient.DownloadFile(url, filenames[index]+".dat"); Console.WriteLine("{0}:{1}", Thread.CurrentThread.ManagedThreadId, url); return webclient; }, (webclient) => { }); }
这样的写法保证了我们能获得足够的 WebClient 实例,同时这些 WebClient 实例彼此隔离仅仅属于各自关联的线程。
虽然 PLINQ 提供了 ThreadLocal<T> 对象来实现类似的功能:
public static void downloadUrl() { var webclient = new ThreadLocal<WebClient>(()=> new WebClient ()); var res = urls .AsParallel() .ForAll( url => { webclient.Value.DownloadFile(url, host[url] +".dat")); Console.WriteLine("{0}:{1}", Thread.CurrentThread.ManagedThreadId, url); }); }
但是请注意:ThreadLocal<T> 相对而言开销更大!
场景五:退出操作 (使用 Parallel.ForEach)
Parallel.ForEach 有个重载声明如下,其中包含一个 ParallelLoopState 对象:
public static ParallelLoopResult ForEach<TSource >( IEnumerable<TSource> source, Action<TSource, ParallelLoopState> body)
ParallelLoopState.Stop() 提供了退出循环的方法,这种方式要比其他两种方法更快。这个方法通知循环不要再启动执行新的迭代,并尽可能快的推出循环。
ParallelLoopState.IsStopped 属性可用来判定其他迭代是否调用了 Stop 方法。
示例:
public static boolean FindAny<T,T>(IEnumerable<T> TSpace, T match) where T: IEqualityComparer<T> { var matchFound = false; Parallel.ForEach(TSpace, (curValue, loopstate) => { if (curValue.Equals(match) ) { matchFound = true; loopstate.Stop(); } }); return matchFound; }
ParallelLoopState.Break() 通知循环继续执行本元素前的迭代,但不执行本元素之后的迭代。最前调用 Break 的起作用,并被记录到 ParallelLoopState.LowestBreakIteration 属性中。这种处理方式通常被应用在一个有序的查找处理中,比如你有一个排序过的数组,你想在其中查找匹配元素的最小 index,那么可以使用以下的代码:
public static int FindLowestIndex<T,T>(IEnumerable<T> TSpace, T match) where T: IEqualityComparer<T> { var loopResult = Parallel.ForEach(source, (curValue, loopState, curIndex) => { if (curValue.Equals(match)) { loopState.Break(); } }); var matchedIndex = loopResult.LowestBreakIteration; return matchedIndex.HasValue ? matchedIndex : -1; }
虽然 PLINQ 也提供了退出的机制(cancellation token),但相对来说退出的时机并没有 Parallel.ForEach 那么及时。
[译]何时使用 Parallel.ForEach,何时使用 PLINQ相关推荐
- 任务并行库(Task Parellel Library)parallel.for parallel.foreach、List、ConcurrentBag 并行集合、线程安全结合
普通的for .foreach 都是顺序依次执行的. C#当中我们一般使用for和foreach执行循环,有时候我们呢的循环结构每一次的迭代需要依赖以前一次的计算或者行为.但是有时候则不需要.如果迭代 ...
- C#并发实战Parallel.ForEach使用
C#并发实战Parallel.ForEach使用 原文:C#并发实战Parallel.ForEach使用 前言:最近给客户开发一个伙食费计算系统,大概需要计算2000个人的伙食.需求是按照员工的预定报 ...
- C# 多线程 Parallel.For 和 For 谁的效率高?那么 Parallel.ForEach 和 ForEach 呢?
还是那句话:十年河东,十年河西,莫欺少年穷. 今天和大家探讨一个问题:Parallel.For 和 For 谁的效率高呢? 从CPU使用方面而言,Parallel.For 属于多线程范畴,可以开辟多个 ...
- C# list删除 另外list里面的元素_C#并发实战Parallel.ForEach使用
前言:最近给客户开发一个伙食费计算系统,大概需要计算2000个人的伙食.需求是按照员工的预定报餐计划对消费记录进行检查,如有未报餐有刷卡或者有报餐没刷卡的要进行一定的金额扣减等一系列规则.一开始我的想 ...
- C# Parellel.For 和 Parallel.ForEach
简介:任务并行库(Task Parellel Library)是BCL的一个类库,极大的简化了并行编程. 使用任务并行库执行循环 C#当中我们一般使用for和foreach执行循环,有时候我们呢的循环 ...
- parallel循环java_Java 8 lambda stream forEach parallel 等循环与Java 7 for each 循环耗时测试...
Java 8 里面的stream 有串行流和并行流之分. 说高级的stream就是那个并行流.下面是那个并行流的简单实现.只要是继承Collection类的都可以这么用. list.stream(). ...
- 第九节:深究并行编程Parallel类中的三大方法 (For、ForEach、Invoke)和几大编程模型(SPM、APM、EAP、TAP)
一. 并行编程 1. 区分串行编程和串行编程 ①. 串行编程:所谓的串行编程就是单线程的作用下,按顺序执行.(典型代表for循环 下面例子从1-100按顺序执行) ②. 并行编程:充分利用多核cpu的 ...
- 深究并行编程Parallel类中的三大方法 (For、ForEach、Invoke)和几大编程模型(SPM、APM、EAP、TAP)
一. 并行编程 1. 区分串行编程和串行编程 ①. 串行编程:所谓的串行编程就是单线程的作用下,按顺序执行.(典型代表for循环 下面例子从1-100按顺序执行) ②. 并行编程:充分利用多核cpu的 ...
- C# 4.0 新特性之并行运算(Parallel)
介绍 C# 4.0 的新特性之并行运算 Parallel.For - for 循环的并行运算 Parallel.ForEach - foreach 循环的并行运算 Parallel.Invoke - ...
最新文章
- UI培训教程分享:Ui设计的细节规范有哪些需要注意?
- 【硅谷牛仔】DropBox CEO --德鲁休斯敦--找到合适自己的圈子,追逐自己感兴趣的事...
- ADO.NET复习总结(5)--工具类SqlHelper 实现登录
- 第4章:介绍python对象类型/4.1 python的核心数据类型/4.7 集合
- 从源码深处体验Spring核心技术--面试中IOC那些鲜为人知的细节
- 《C++ Primer》13.1.2节练习
- mysql连接代替子查询_MySQL优化之使用连接(join)代替子查询
- python连接服务器sql server_python – 无法通过pymssql连接到SQL Server
- 产品经理学技术之数据结构
- birt脚本for循环语句_python循环语句(while amp; for)
- HAProxy的安装和部署
- php 获取图片、swf的尺寸大小
- 刘宇凡:新型鸡汤如何击溃传统鸡汤?
- Unity+Kinect 开发脚本介绍
- 抖音App已正式更名,短视频在名称中消失
- 解析html,批量下载图片(java实现)
- unity开发xbox手柄 驱动坑
- 芯片对蓝牙耳机影响有多大?南卡N2s发布,搭载高通3040芯片
- VitalTools 唯她CAD工具集
- vue3项目打包优化三步走