1、简介
关于并行Linq,Ms官方叫做并行语言集成(PLINQ)查询,其实本质就是Linq的多线程版本,常规的Linq是单线程的,也就是同步的过程处理完所有的查询.如果你的Linq查询足够简单,而且耗时短,那么建议你使用Linq,但是如果你的查询比较耗时,而且很复杂,且不涉及多线程争用问题,那么可以使用PLinq技术,让多个线程参与到查询中来,有效的利用CPU资源.这样你的代码能从中获得最大的收益.判断什么时候使用PLINQ,什么时候使用Linq?这需要你自己去实践,因为不同的环境,产生的效果不一样,因为我前面的随笔中介绍了,多线程(Task,因为Parallel是基于Task的)本身的开销,CPU的上下文切换,都是影响的因素.可能你使用PLINQ执行一个复杂的查询,本地的运行速度很快,但是放到服务器上去反而变慢了.所以使用还是需要慎重.
2、代码结构简介
(1)、基本Api介绍
那么如何使用PLINQ呢?所有的PLINQ的Api都在System.Linq.ParallelEnumerable类下面,Api几乎和Linq一样,因为内容太多,这里就不截图了.MS几乎将常规的LINQ所有的Api都实现了一个并行版本.所有的方法都是ParallelQuery<TSource>类型的扩展，如下:
所有如果你有一个常规集合需要进行并行查询,那么你需要将该集合转换成ParallelQuery<TSource>类型,MS提供了转换方法,如下:
主要是红框中的两个,一个泛型版本,一个非泛型版本,本文主要介绍这两个,其余的稍微介绍下.
:
调用这个方法,它将执行并行查询切换为同步查询,但是不常用.
调用这个方法,线程将成组处理数据,然后将数据项合并回去,同时保持顺序,会产生一定的性能损耗.
注:如果你调用的不是对数据源进行排序的方法,那么它们的并行处理结果是无序的,每次都会变,但是如果你希望有序之后变无序,可以调用
但是没有人会这么干!
(2)、构造可取消的PLINQ查询
接受一个CancellationToken参数,支持显示取消.
(3)、构造线程数限制的PLINQ查询
接受一个最大的可分配线程数参数,一般小于内核数.
(4)、构造一个强制以并行方式执行的PLINQ查询
因为并不并行,是PLINQ内部机制决定的,所以可能你的查询过于简单,它会以并行的方式处理,所以如果你需要强制它以并行方式执行可以调用
并给后面的枚举设置
(5)、指定多个线程处理完数据源后已何种方式合并处理完的数据项
指定不同的枚举项,会对性能产生影响。建议你每个都是试一试,就知道哪个更适合你的接口.一般默认的就够了.因为PLINQ调度内核的方式很复杂,所以这里不多介绍.
3、实战
将一个模块程序集中的所有查询接口和查询实体放到一个实例中,并返回.
User模块的代码结构如下:
class ParallelLinqStudy { static void Main(string[] args) { var modules=Register("User"); Console.ReadKey(); } static object lockObjOne = new object(); static object lockObjTwo = new object(); static ModultInfo Register(params string[] assembies) { var moduleInfo = new ModultInfo(); assembies.ForEach(assembly => { var ass=Assembly.Load(assembly); var allTypes = ass.GetTypes().AsParallel(); //遍历传入程序集,将所有实现了IQuery接口的接口类型,并将其在控制台上输出 allTypes.Where(w => w.ImplInterfance ()).Where(w => w.IsInterface && w.Name!= "IQuery").ForEach(f => { lock (lockObjOne) { moduleInfo.IQueries.Add(f); } allTypes.Where(w => f.IsAssignableFrom(w) && !w.IsInterface).ForEach(type => { lock (lockObjTwo) { moduleInfo.Queries.Add(type); } }); }); }); return moduleInfo; } } class ModultInfo { public List IQueries { get; set; } = new List (); public List Queries { get; set; } = new List (); } /// /// Type扩展 /// static class TypeExtension { /// /// 判断传入类型type是否实现了Interface接口 /// /// (this Type type) { //接口实例是可以分配给实现类型的,而实例是不可以分配给接口实例的 return typeof(Interface).IsAssignableFrom(type); } } /// /// Linq扩展 /// static class LinqExtension { public static void ForEach (this IEnumerable enumerators, Action action) { foreach (var item in enumerators) { action(item); } } } 
上面的代码给List加了锁,因为它是线程不安全的,具体请参考我的这篇
ok,现在拿到了所有的Query接口和Query实体,如果后续需要对这两个集合进行后续的只读操作,可以使用Parallel(参考我前面的)进行并行的只读操作,如果操作很耗时,或者很复杂.也可以将集合转换为ParallelQuery<TSource>类型,并使用
方法进行后续的并行操作.代码如下:
static void Main(string[] args) { var modules = Register("User"); modules.IQueries.AsParallel().ForAll(iQuery => { //执行一个不带返回值的操作 }); Console.ReadKey(); }