C# 多线程八之并行Linq(ParallelEnumerable)
C# 多线程八之并行Linq(ParallelEnumerable)
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<IQuery>()).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<Type> IQueries { get; set; } = new List<Type>();
public List<Type> Queries { get; set; } = new List<Type>();
}
/// <summary>
/// Type扩展
/// </summary>
static class TypeExtension
{
/// <summary>
/// 判断传入类型type是否实现了Interface接口
/// </summary>
/// <typeparam name="Interface"></typeparam>
/// <param name="type"></param>
/// <returns></returns>
public static bool ImplInterfance<Interface>(this Type type)
{
//接口实例是可以分配给实现类型的,而实例是不可以分配给接口实例的
return typeof(Interface).IsAssignableFrom(type);
}
}
/// <summary>
/// Linq扩展
/// </summary>
static class LinqExtension
{
public static void ForEach<T>(this IEnumerable<T> enumerators, Action<T> 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();
}