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();

        }

巴特西

C# 多线程八之并行Linq(ParallelEnumerable)

最新文章

热门文章