C#並行編程中的Parallel.Invoke

時間 2020-06-29

標籤 c# 並行編程 parallel.invoke parallel invoke 欄目 C# 简体版

原文原文鏈接

1、基礎知識html

並行編程：並行編程是指軟件開發的代碼，它能在同一時間執行多個計算任務，提升執行效率和性能一種編程方式，屬於多線程編程範疇。因此咱們在設計過程當中通常會將不少任務劃分紅若干個互相獨立子任務，這些任務不考慮互相的依賴和順序。這樣咱們就可使用很好的使用並行編程。可是咱們都知道多核處理器的並行設計使用共享內存，若是沒有考慮併發問題，就會有不少異常和達不到咱們預期的效果。不過還好NET Framework4.0引入了Task Parallel Library（TPL）實現了基於任務設計而不用處理重複複雜的線程的並行開發框架。它支持數據並行，任務並行與流水線。核心主要是Task，可是通常簡單的並行咱們能夠利用Parallel提供的靜態類以下三個方法。
編程

Parallel.Invoke 對給定任務實現並行開發多線程

Parallel.For 對固定數目的任務提供循環迭代並行開發併發

parallel.Foreach 對固定數目的任務提供循環迭代並行開發框架

注意：全部的並行開發不是簡單的覺得只要將For或者Foreach換成Parallel.For與Parallel.Foreach這樣簡單。異步

PS：從簡單的Invoke開始逐步深刻探討並行開發的主要知識點，也對本身學習過程當中的積累作個總結，其中參考了博客園中的其餘優秀博文異步編程

滴答的雨異步編程：輕量級線程同步基元對象post

首先感謝您，在我學習並行開發過程當中，您的博文對我幫助很大。性能

2、Parallel.Invoke在並行中的使用學習

首先咱們來看看它的兩個重載方法：

　　public static void Invoke(params Action[] actions);  
　　public static void Invoke(ParallelOptions parallelOptions, params Action[] actions);

Invoke主要接受params的委託actions，好比咱們要同時執行三個任務，咱們能夠這樣利用

  方式一
       Parallel.Invoke(() => Task1(), () => Task2(), () => Task3());
  方式二
       Parallel.Invoke(Task1, Task2, Task3);
  方式三
       Parallel.Invoke(
                () =>
                {
                    Task1();
                },
                Task2,
                delegate () { Task3(); console.write('do someting!');});

這樣Invoke就簡單實現了Task1，Task2，Task3的並行開發。下面咱們用實例來講明他們的執行規則。以及兩個重載方法的使用。

三、Demo

一、 Demo 1：

public class ParallelInvoke
    {
        /// <summary>
        /// Invoke方式一 action
        /// </summary>
        public  void Client1()
        {
            Stopwatch stopWatch = new Stopwatch();
           
            Console.WriteLine("主線程：{0}線程ID ： {1}；開始", "Client1", Thread.CurrentThread.ManagedThreadId);
            stopWatch.Start();
            Parallel.Invoke(() => Task1("task1"), () => Task2("task2"), () => Task3("task3"));
            stopWatch.Stop();
            Console.WriteLine("主線程：{0}線程ID ： {1}；結束,共用時{2}ms", "Client1", Thread.CurrentThread.ManagedThreadId, stopWatch.ElapsedMilliseconds);
        }

        private void Task1(string data)
        {
            Thread.Sleep(5000);
            Console.WriteLine("任務名：{0}線程ID ： {1}", data, Thread.CurrentThread.ManagedThreadId);
        }

        private void Task2(string data)
        {
            Console.WriteLine("任務名：{0}線程ID ： {1}", data, Thread.CurrentThread.ManagedThreadId);
        }

        private void Task3(string data)
        {
            Console.WriteLine("任務名：{0}線程ID ： {1}", data, Thread.CurrentThread.ManagedThreadId);
        }
｝

執行運行後結果：

咱們看到Invoke 執行Task三個方法主要有如下幾個特色：

一、沒有固定的順序，每一個Task多是不一樣的線程去執行，也多是相同的；

二、主線程必須等Invoke中的全部方法執行完成後返回才繼續向下執行；這樣對咱們之後設計並行的時候，要考慮每一個Task任務儘量差很少，若是相差很大，好比一個時間很是長，其餘都比較短，這樣一個線程可能會影響整個任務的性能。這點很是重要

三、這個很是簡單就實現了並行，不用咱們考慮線程問題。主要Framework已經爲咱們控制好線程池的問題。

ps：若是其中有一個異常怎麼辦？帶作這個問題修改了增長了一個Task4.

二、 Demo2

    public class ParallelInvoke
    {
        /// <summary>
        /// Invoke方式一 action
        /// </summary>
        public  void Client1()
        {
            Stopwatch stopWatch = new Stopwatch();
           
            Console.WriteLine("主線程：{0}線程ID ： {1}；開始", "Client1", Thread.CurrentThread.ManagedThreadId);
            stopWatch.Start();

            try
            {
                Parallel.Invoke(() => Task1("task1"), () => Task2("task2"), () => Task3("task3"), delegate () { throw new Exception("我這裏發送了異常"); });
            }
            catch (AggregateException ae)
            {
                foreach (var ex in ae.InnerExceptions)
                    Console.WriteLine(ex.Message);
            }
          
            stopWatch.Stop();
            Console.WriteLine("主線程：{0}線程ID ： {1}；結束,共用時{2}ms", "Client1", Thread.CurrentThread.ManagedThreadId, stopWatch.ElapsedMilliseconds);
        }
   ｝

主要看 delegate() { throw new Exception("我這裏發送了異常");} 增長了這個委託Task3. 而後咱們看結果：

這裏咱們發現即便有異常程序也會完成執行，並且不會影響其餘Task的執行。

三、demo3 重載方法ParallelOptions 的使用。

理解ParallelOptions建議你們異步編程：輕量級線程同步基元對象講的很是詳細。

主要理解兩個參數：

CancellationToken 控制線程的取消
MaxDegreeOfParallelism 設置最大的線程數，有時候可能會跑遍全部的內核，爲了提升其餘應用程序的穩定性，就要限制參與的內核

下面從代碼上看效果如何？

 public class ParallelInvoke
    {
 　　　　// 定義CancellationTokenSource 控制取消
        readonly CancellationTokenSource _cts = new CancellationTokenSource();

        /// <summary>
        /// Invoke方式一 action
        /// </summary>
        public  void Client1()
        {
             Console.WriteLine("主線程：{0}線程ID ： {1}；開始{2}", "Client3", Thread.CurrentThread.ManagedThreadId, DateTime.Now);
            
            var po = new ParallelOptions
            {
                CancellationToken = _cts.Token, // 控制線程取消
                MaxDegreeOfParallelism = 3  // 設置最大的線程數3，仔細觀察線程ID變化
            };
           
            Parallel.Invoke(po, () => Task1("task1"), ()=>Task5(po), Task6);
            
            Console.WriteLine("主線程：{0}線程ID ： {1}；結束{2}", "Client3", Thread.CurrentThread.ManagedThreadId, DateTime.Now);
        }

        private void Task1(string data)
        {
            Thread.Sleep(5000);
            Console.WriteLine("任務名：{0}線程ID ： {1}", data, Thread.CurrentThread.ManagedThreadId);
        }

        // 打印數字
        private void Task5(ParallelOptions po)
        {
            Console.WriteLine("進入Task5線程ID ： {0}", Thread.CurrentThread.ManagedThreadId);
            int i = 0;
          while (i < 100)
            {
                // 判斷是否已經取消
                if (po.CancellationToken.IsCancellationRequested)
                {
                    Console.WriteLine("已經被取消。");
                    return;
                }

                Thread.Sleep(100);
                Console.Write(i + " ");
                Interlocked.Increment(ref i);
            }

        }

        /// <summary>
        /// 10秒後取消
        /// </summary>
        private void Task6()
        {
            Console.WriteLine("進入取消任務，Task6線程ID ： {0}", Thread.CurrentThread.ManagedThreadId);
            Thread.Sleep(1000 * 10);
            _cts.Cancel();
            Console.WriteLine("發起取消請求...........");
        }
｝