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