快捷導(dǎo)航

詳解c# 并行計算

更新時間：2020年12月18日 08:38:44 作者：jett010

本文主要介紹了并行計算的簡單使用，并行循環(huán)的中斷和跳出、并行循環(huán)中為數(shù)組/集合添加項、返回集合運算結(jié)果/含有局部變量的并行循環(huán)、、PLinq（Linq的并行計算）等相關(guān)內(nèi)容。

并行計算部分

沿用微軟的寫法，System.Threading.Tasks.::.Parallel類，提供對并行循環(huán)和區(qū)域的支持。我們會用到的方法有For，F(xiàn)orEach，Invoke。

一、簡單使用

首先我們初始化一個List用于循環(huán)，這里我們循環(huán)10次。（后面的代碼都會按這個標(biāo)準(zhǔn)進行循環(huán)）

            Program.Data = new List<int>();
            for (int i = 0; i < 10; i++)
            {
                Data.Add(i);
            }

下面我們定義4個方法，分別為for，foreach，并行For，并行ForEach。并測試他們的運行時長。

        /// <summary>
        /// 是否顯示執(zhí)行過程
        /// </summary>
        public bool ShowProcessExecution = false;
        /// <summary>
        /// 這是普通循環(huán)for
        /// </summary>
        private void Demo1()
        {
            List<int> data = Program.Data;
            DateTime dt1 = DateTime.Now;
            for (int i = 0; i < data.Count; i++)
            {
                Thread.Sleep(500);
                if (ShowProcessExecution)
                    Console.WriteLine(data[i]);
            }
            DateTime dt2 = DateTime.Now;
            Console.WriteLine("普通循環(huán)For運行時長：{0}毫秒。", (dt2 - dt1).TotalMilliseconds);
        }
        /// <summary>
        /// 這是普通循環(huán)foreach
        /// </summary>
        private void Demo2()
        {
            List<int> data = Program.Data;
            DateTime dt1 = DateTime.Now;
            foreach (var i in data)
            {
                Thread.Sleep(500);
                if (ShowProcessExecution)
                    Console.WriteLine(i);
            }
            DateTime dt2 = DateTime.Now;
            Console.WriteLine("普通循環(huán)For運行時長：{0}毫秒。", (dt2 - dt1).TotalMilliseconds);
        }
        /// <summary>
        /// 這是并行計算For
        /// </summary>
        private void Demo3()
        {
            List<int> data = Program.Data;
            DateTime dt1 = DateTime.Now;
            Parallel.For(0, data.Count, (i) =>
            {
                Thread.Sleep(500);
                if (ShowProcessExecution)
                    Console.WriteLine(data[i]);
            });
            DateTime dt2 = DateTime.Now;
            Console.WriteLine("并行運算For運行時長：{0}毫秒。", (dt2 - dt1).TotalMilliseconds);
        }
        /// <summary>
        /// 這是并行計算ForEach
        /// </summary>
        private void Demo4()
        {
            List<int> data = Program.Data;
            DateTime dt1 = DateTime.Now;
            Parallel.ForEach(data, (i) =>
            {
                Thread.Sleep(500);
                if (ShowProcessExecution)
                    Console.WriteLine(i);
            });
            DateTime dt2 = DateTime.Now;
            Console.WriteLine("并行運算ForEach運行時長：{0}毫秒。", (dt2 - dt1).TotalMilliseconds);
        }

下面是運行結(jié)果：

這里我們可以看出并行循環(huán)在執(zhí)行效率上的優(yōu)勢了。

結(jié)論1：在對一個數(shù)組內(nèi)的每一個項做單獨處理時，完全可以選擇并行循環(huán)的方式來提升執(zhí)行效率。

原理1：并行計算的線程開啟是緩步開啟的，線程數(shù)量1,2,4,8緩步提升。（不詳，PLinq最多64個線程，可能這也是64）

二、并行循環(huán)的中斷和跳出

當(dāng)在進行循環(huán)時，偶爾會需要中斷循環(huán)或跳出循環(huán)。下面是兩種跳出循環(huán)的方法Stop和Break，LoopState是循環(huán)狀態(tài)的參數(shù)。

        /// <summary>
        /// 中斷Stop
        /// </summary>
        private void Demo5()
        {
            List<int> data = Program.Data;
            Parallel.For(0, data.Count, (i, LoopState) =>
            {
                if (data[i] > 5)
                    LoopState.Stop();
                Thread.Sleep(500);
                Console.WriteLine(data[i]);
            });
            Console.WriteLine("Stop執(zhí)行結(jié)束。");
        }
        /// <summary>
        /// 中斷Break
        /// </summary>
        private void Demo6()
        {
            List<int> data = Program.Data;
            Parallel.ForEach(data, (i, LoopState) =>
            {
                if (i > 5)
                    LoopState.Break();
                Thread.Sleep(500);
                Console.WriteLine(i);
            });
            Console.WriteLine("Break執(zhí)行結(jié)束。");
        }

執(zhí)行結(jié)果如下：

結(jié)論2：使用Stop會立即停止循環(huán)，使用Break會執(zhí)行完畢所有符合條件的項。

三、并行循環(huán)中為數(shù)組/集合添加項

上面的應(yīng)用場景其實并不是非常多見，畢竟只是為了遍歷一個數(shù)組內(nèi)的資源，我們更多的時候是為了遍歷資源，找到我們所需要的。那么請繼續(xù)看。

下面是我們一般會想到的寫法：

        private void Demo7()
        {
            List<int> data = new List<int>();
            Parallel.For(0, Program.Data.Count, (i) =>
            {
                if (Program.Data[i] % 2 == 0)
                    data.Add(Program.Data[i]);
            });
            Console.WriteLine("執(zhí)行完成For.");
        }
        private void Demo8()
        {
            List<int> data = new List<int>();
            Parallel.ForEach(Program.Data, (i) =>
            {
                if (Program.Data[i] % 2 == 0)
                    data.Add(Program.Data[i]);
            });
            Console.WriteLine("執(zhí)行完成ForEach.");
        }

看起來應(yīng)該是沒有問題的，但是我們多次運行后會發(fā)現(xiàn)，偶爾會出現(xiàn)錯誤如下：

這是因為List是非線程安全的類，我們需要使用System.Collections.Concurrent命名空間下的類型來用于并行循環(huán)體內(nèi)。

類	說明
BlockingCollection<T>	為實現(xiàn) IProducerConsumerCollection<T> 的線程安全集合提供阻止和限制功能。
ConcurrentBag<T>	表示對象的線程安全的無序集合。
ConcurrentDictionary<TKey, TValue>	表示可由多個線程同時訪問的鍵值對的線程安全集合。
ConcurrentQueue<T>	表示線程安全的先進先出 (FIFO) 集合。
ConcurrentStack<T>	表示線程安全的后進先出 (LIFO) 集合。
OrderablePartitioner<TSource>	表示將一個可排序數(shù)據(jù)源拆分成多個分區(qū)的特定方式。
Partitioner	提供針對數(shù)組、列表和可枚舉項的常見分區(qū)策略。
Partitioner<TSource>	表示將一個數(shù)據(jù)源拆分成多個分區(qū)的特定方式。

那么我們上面的代碼可以修改為，加了了ConcurrentQueue和ConcurrentStack的最基本的操作。

        /// <summary>
        /// 并行循環(huán)操作集合類,集合內(nèi)只取5個對象
        /// </summary>
        private void Demo7()
        {
            ConcurrentQueue<int> data = new ConcurrentQueue<int>();
            Parallel.For(0, Program.Data.Count, (i) =>
            {
                if (Program.Data[i] % 2 == 0)
                    data.Enqueue(Program.Data[i]);//將對象加入到隊列末尾
            });
            int R;
            while (data.TryDequeue(out R))//返回隊列中開始處的對象
            {
                Console.WriteLine(R);
            }
            Console.WriteLine("執(zhí)行完成For.");
        }
        /// <summary>
        /// 并行循環(huán)操作集合類
        /// </summary>
        private void Demo8()
        {
            ConcurrentStack<int> data = new ConcurrentStack<int>();
            Parallel.ForEach(Program.Data, (i) =>
            {
                if (Program.Data[i] % 2 == 0)
                    data.Push(Program.Data[i]);//將對象壓入棧中
            });
            int R;
            while (data.TryPop(out R))//彈出棧頂對象
            {
                Console.WriteLine(R);
            }
            Console.WriteLine("執(zhí)行完成ForEach.");
        }

ok，這里返回一個序列的問題也解決了。

結(jié)論3:在并行循環(huán)內(nèi)重復(fù)操作的對象，必須要是thread-safe(線程安全)的。集合類的線程安全對象全部在System.Collections.Concurrent命名空間下。

四、返回集合運算結(jié)果/含有局部變量的并行循環(huán)

使用循環(huán)的時候經(jīng)常也會用到迭代，那么在并行循環(huán)中叫做含有局部變量的循環(huán) 。下面的代碼中詳細的解釋，這里就不啰嗦了。

        /// <summary>
        /// 具有線程局部變量的For循環(huán)
        /// </summary>
        private void Demo9()
        {
            List<int> data = Program.Data;
            long total = 0;
            //這里定義返回值為long類型方便下面各個參數(shù)的解釋
            Parallel.For<long>(0,           // For循環(huán)的起點
                data.Count,                 // For循環(huán)的終點
                () => 0,                    // 初始化局部變量的方法(long)，既為下面的subtotal的初值
                (i, LoopState, subtotal) => // 為每個迭代調(diào)用一次的委托，i是當(dāng)前索引，LoopState是循環(huán)狀態(tài)，subtotal為局部變量名
                {
                    subtotal += data[i];    // 修改局部變量
                    return subtotal;        // 傳遞參數(shù)給下一個迭代
                },
                (finalResult) => Interlocked.Add(ref total, finalResult) //對每個線程結(jié)果執(zhí)行的最后操作，這里是將所有的結(jié)果相加
                );
            Console.WriteLine(total);
        }
        /// <summary>
        /// 具有線程局部變量的ForEach循環(huán)
        /// </summary>
        private void Demo10()
        {
            List<int> data = Program.Data;
            long total = 0;
            Parallel.ForEach<int, long>(data, // 要循環(huán)的集合對象
                () => 0,                      // 初始化局部變量的方法(long)，既為下面的subtotal的初值
                (i, LoopState, subtotal) =>   // 為每個迭代調(diào)用一次的委托，i是當(dāng)前元素，LoopState是循環(huán)狀態(tài)，subtotal為局部變量名
                {
                    subtotal += i;            // 修改局部變量
                    return subtotal;          // 傳遞參數(shù)給下一個迭代
                },
                (finalResult) => Interlocked.Add(ref total, finalResult) //對每個線程結(jié)果執(zhí)行的最后操作，這里是將所有的結(jié)果相加
                );
            Console.WriteLine(total);
        }

結(jié)論4:并行循環(huán)中的迭代，確實很傷人。代碼太難理解了。

五、PLinq（Linq的并行計算）

上面介紹完了For和ForEach的并行計算盛宴，微軟也沒忘記在Linq中加入并行計算。下面介紹Linq中的并行計算。

4.0中在System.Linq命名空間下加入了下面幾個新的類：

類	說明
ParallelEnumerable	提供一組用于查詢實現(xiàn) ParallelQuery{TSource} 的對象的方法。這是 Enumerable 的并行等效項。
ParallelQuery	表示并行序列。
ParallelQuery<TSource>	表示并行序列。

原理2：PLinq最多會開啟64個線程

原理3：PLinq會自己判斷是否可以進行并行計算，如果不行則會以順序模式運行。

原理4：PLinq會在昂貴的并行算法或成本較低的順序算法之間進行選擇，默認情況下它選擇順序算法。

在ParallelEnumerable中提供的并行化的方法

ParallelEnumerable 運算符	說明
AsParallel()	PLINQ 的入口點。指定如果可能，應(yīng)并行化查詢的其余部分。
AsSequential()	指定查詢的其余部分應(yīng)像非并行 LINQ 查詢一樣按順序運行。
AsOrdered()	指定 PLINQ 應(yīng)保留查詢的其余部分的源序列排序，直到例如通過使用 orderby 子句更改排序為止。
AsUnordered()	指定查詢的其余部分的 PLINQ 不需要保留源序列的排序。
WithCancellation()	指定 PLINQ 應(yīng)定期監(jiān)視請求取消時提供的取消標(biāo)記和取消執(zhí)行的狀態(tài)。
WithDegreeOfParallelism()	指定 PLINQ 應(yīng)當(dāng)用來并行化查詢的處理器的最大數(shù)目。
WithMergeOptions()	提供有關(guān) PLINQ 應(yīng)當(dāng)如何（如果可能）將并行結(jié)果合并回到使用線程上的一個序列的提示。
WithExecutionMode()	指定 PLINQ 應(yīng)當(dāng)如何并行化查詢（即使默認行為是按順序運行查詢）。
ForAll()	多線程枚舉方法，與循環(huán)訪問查詢結(jié)果不同，它允許在不首先合并回到使用者線程的情況下并行處理結(jié)果。
Aggregate() 重載	對于 PLINQ 唯一的重載，它啟用對線程本地分區(qū)的中間聚合以及一個用于合并所有分區(qū)結(jié)果的最終聚合函數(shù)。

下面是PLinq的簡單代碼

        /// <summary>
        /// PLinq簡介
        /// </summary>
        private void Demo11()
        {
            var source = Enumerable.Range(1, 10000);
            //查詢結(jié)果按source中的順序排序
            var evenNums = from num in source.AsParallel().AsOrdered()
                       where num % 2 == 0
                       select num;
            //ForAll的使用
            ConcurrentBag<int> concurrentBag = new ConcurrentBag<int>();
            var query = from num in source.AsParallel()
                        where num % 10 == 0
                        select num;
            query.ForAll((e) => concurrentBag.Add(e * e));
        }

上面代碼中使用了ForAll，F(xiàn)orAll和foreach的區(qū)別如下：