C#並行編程之Parallel的使用

前言：在C#的System.Threading.Tasks 命名空間中有一個靜態的並行類：Parallel，封裝了Task的使用，對於執行大量任務提供了非常簡便的操作。下面對他的使用進行介紹。

 

本篇內容：

1.1、Parallel.For 使用
1.2、Parallel.ForEach 使用
1.3、Parallel.Invoke 使用
1.4、ParallelOptions 選項配置
1.5、ParallelLoopResult 執行結果
1.6、ParallelLoopState 提前結束
1.7、Parallel的使用場景分析

 

1.1、Parallel.For 使用

首先創建一個控制台程序，本案例使用的是.net core 3.1，引入命名空間 using System.Threading。假設某個操作需要執行10次，從0到9代碼如下：

        static void ParallelFor(int num)
        {
            Console.WriteLine($"ParallelFor執行 {num} 次");
            var list = new List(num);
            ParallelLoopResult result = Parallel.For(0, num, i =>
              {
                  list.Add(new Product { Id = i, Name = "TestName" });
                  Console.WriteLine($"Task Id：{Task.CurrentId},Thread: {Thread.CurrentThread.ManagedThreadId}");
                  Thread.Sleep(10);
              });
        }

 執行結果如下：

 從打印信息可以看出，任務Id和線程都是無序的，在使用時需要注意。

Parallel.For 還提供了很多重載本版：

 我們看一下帶ParallelLoopState 參數的一個重載版本：ParallelLoopResult For(int fromInclusive, int toExclusive, Action<int, ParallelLoopState> body)

測試代碼：

        /// <summary>
        /// 提前終止
        /// </summary>
        static void ParallelForAsyncAbort()
        {
            ParallelLoopResult result =
                Parallel.For(10, 100, async (int index, ParallelLoopState pls) =>
                {
                    Console.WriteLine($"index:{index} task:{Task.CurrentId},Thread:{Thread.CurrentThread.ManagedThreadId}");
                    await Task.Delay(10);
                    if (index > 30)
                        pls.Break();
                });
            Console.WriteLine($"Is completed：{result.IsCompleted} LowestBreakIteration：{result.LowestBreakIteration}");
        }

執行結果：

 任務提前結束了，最小執行Break方法的索引為19

帶參數：ParallelOptions的重載版本：

/// <summary>
        /// 執行500毫秒后取消
        /// </summary>
        static void ParalletForCancel()
        {
            var cts = new CancellationTokenSource();
            cts.Token.Register(() => { Console.WriteLine($"*** token canceled"); }
            );
            // send a cancel after 500ms
            cts.CancelAfter(500);
            try
            {
                ParallelLoopResult result =
                     Parallel.For(0, 100, new ParallelOptions()
                     {
                         CancellationToken = cts.Token,
                     }, x =>
                     {
                         Console.WriteLine($"loop {x} started");
                         int sum = 0;
                         for (int i = 0; i < 100; i++)
                         {
                             Thread.Sleep(2);
                             sum += i;
                         }
                         Console.WriteLine($"loop {x} finished");
                     });

            }
            catch (OperationCanceledException ex)
            {
                Console.WriteLine(ex.Message);
            }
        }

任務執行一段時間后取消了

 

 

 1.2、Parallel.ForEach 使用

ForEach方法可用於對集合，數組，或枚舉進行循環操作，下面進行簡單使用：

        //簡單使用
        static void ParallelForEach()
        {
            string[] data = { "zero", "one", "two", "three", "four", "five", "six", "seven", "eight", "nine", "ten" };
            ParallelLoopResult result =
                    Parallel.ForEach(data, a =>
                    {
                        Console.WriteLine(a);
                    });
        }

執行結果：

 請注意循環的執行是無序的，我們打印出執行順序：

   //帶索引的循環操作
        static void ParallelForEach2()
        {
            string[] data = { "zero", "one", "two", "three", "four", "five", "six", "seven", "eight", "nine", "ten" };
            ParallelLoopResult result =
                    Parallel.ForEach<string>(data, (str, psl, index) =>
                    {
                        Console.WriteLine($"str：{str} index：{index}");
                    });
        }

執行結果：

 ForEach同樣支持提前結束和取消操作：

 1.3、Parallel.Invoke 使用

Invoke主要用於操作（任務）並行，能同時執行多個操作，並盡可能的同時執行。

簡單使用：

        static void ParallelInvoke()
        {
            Parallel.Invoke(Foo, Bar);
        }

        static void Foo()
        {
            Console.WriteLine("Foo");
        }

        static void Bar()
        {
            Console.WriteLine("Bar");
        }

大於10個操作：

        static void ParallelInvoke2()
        {
            Action action = () =>
            {
                Console.WriteLine($"Thread Id:{Thread.CurrentThread.ManagedThreadId}");
            };
            Parallel.Invoke(action, action, action, action, action, action, action, action, action, action, action);
            Console.WriteLine("Parallel.Invoke 執行完畢");
          }

如果任務不超過10個，Invoke內部會使用Task.Factory.StartNew 創建任務，效率不高，不如直接使用Task。

 1.4、ParallelOptions 選項配置

 ParallelOptions是一個選項配置，有三個屬性：

 1.4.1、CancellationToken-定義取消令牌，處理任務被取消后的一些操作

1.4.2、MaxDegreeOfParallelism-設置最大並發限制，默認-1

1.4.3、TaskScheduler 指定任務調度器

 1.5、ParallelLoopResult 執行結果

ParallelLoopResult，並發循環結果，有兩個屬性：

 

IsCompleted-任務是否執行完

LowestBreakIteration-調用Break方法的最小任務的索引

 

1.6、ParallelLoopState 提前結束

ParallelLoopState 用於提前結束循環操作，比如搜索算法，已找到結果提前結束查詢。

有兩個方法：

 

Break: 告知 Parallel 循環應在系統方便的時候盡早停止執行當前迭代之外的迭代

Stop：告知 Parallel 循環應在系統方便的時候盡早停止執行。

如果循環之外還有需要執行的代碼則用Break,否則使用Stop

 

 1.7、Parallel的使用場景分析

 1.7.1、Parallel.Invoke 使用特點：

     1、如果操作小於10個，使用Task.Factory.StartNew 或者Task.Run 效率更高

     2、適合用於執行大量操作且無需返回結果的場景

1.7.2、Parallel.For 使用特點：

     1、帶索引的大量循環操作

1.7.3、Parallel.ForEach 使用特點：

    1、大數據集（數組，集合，枚舉集）的循環執行

1.7.4、注意事項：

    1、循環操作是無序的，如果需要順序直接請使用同步執行

    2、如果涉及操作共享變量請使用線程同步鎖

    3、如果是簡單、量大且無等待的操作可能並不適用，同步執行可能更快

    4、注意錯誤的處理，如果是帶數據庫的操作請注意事務的使用

    5、個人測試，Parallel.ForEach 的使用效率比Parallel.For更高

性能測試代碼如下：

        #region 性能測試
        private static void TestPerformance()
        {
            int num = 10;
            Console.WriteLine($"測試執行：{num}次");
            Console.WriteLine();
            Stopwatch stopwatch = new Stopwatch();
            stopwatch.Start();
            ParallelFor(num);
            stopwatch.Stop();
            Console.WriteLine($"耗時：{stopwatch.ElapsedMilliseconds}");
            Console.WriteLine();
            stopwatch.Restart();
            ParallelForEach(num);
            stopwatch.Stop();
            Console.WriteLine($"耗時：{stopwatch.ElapsedMilliseconds}");
            Console.WriteLine();
            stopwatch.Restart();
            Sync(num);
            stopwatch.Stop();
            Console.WriteLine($"耗時：{stopwatch.ElapsedMilliseconds}");
            Console.WriteLine();
            stopwatch.Restart();
            TaskTest(num);
            stopwatch.Stop();
            Console.WriteLine($"耗時：{stopwatch.ElapsedMilliseconds}");
        }

        static void ParallelFor(int num)
        {
            Console.WriteLine($"ParallelFor執行 {num} 次");
            var list = new List<Product>(num);
            ParallelLoopResult result = Parallel.For(0, num, i =>
            {
                list.Add(new Product { Id = i, Name = "TestName" });
                //去掉Thread的代碼模擬簡單業務操作
                Thread.Sleep(10);
            });
        }

        static void Sync(int num)
        {
            Console.WriteLine($"同步執行 {num} 次");
            var list = new List<Product>(num);
            for (int i = 0; i < num; i++)
            {
                list.Add(new Product { Id = i, Name = "TestName" });
                Thread.Sleep(10);
            }
        }

        static void ParallelForEach(int num)
        {
            string[] datas = new string[num];
            Console.WriteLine($"ParallelForEach執行 {num} 次");
            var list = new List<Product>(num);
            Parallel.ForEach(datas, (s, pls, i) =>
            {
                list.Add(new Product { Id = (int)i, Name = "TestName" });
                Thread.Sleep(10);
            });
        }
        static void TaskTest(int num)
        {
            Console.WriteLine($"Task 執行 {num} 次");
            var list = new List<Product>(num);
            while (num > 0)
            {
                Task.Run(() =>
                {
                    list.Add(new Product { Id = num, Name = "TestName" });
                });
                Thread.Sleep(10);
                num--;
            }
        }

        #endregion

簡單任務性能測試：

 

 

 

 

 

 

 復雜任務性能測試模擬：

 

 

 

 

 以上是我對Parallel的學習和使用經驗總結，歡迎大家一起交流和學習。

參考：《C#高級編程第4版》、微軟官網