编程语言
首页 > 编程语言> > 多线程---Parallel(并行编程)

多线程---Parallel(并行编程)

作者:互联网

1.基本介绍

Parallel类是对线程的一个很好的抽象。该类位于System.Threading.Tasks命名空间中,提供了数据和任务并行性。

2.Parallel.Invoke 主要用于任务的并行

这个函数的功能和Task有些相似,就是并发执行一系列任务,然后等待所有完成。和Task比起来,省略了Task.WaitAll这一步,自然也缺少了Task的相关管理功能。它有两种形式:
  Parallel.Invoke( params Action[] actions);
  Parallel.Invoke(Action[] actions,TaskManager manager,TaskCreationOptions options);

      static void Main()
        {
            try
            {
                Parallel.Invoke(
                    BasicAction,    
                    () =>            
                    {
                        Console.WriteLine("Method=beta, Thread={0}", Thread.CurrentThread.ManagedThreadId);
                    },
                    delegate ()        
                    {
                        Console.WriteLine("Method=gamma, Thread={0}", Thread.CurrentThread.ManagedThreadId);
                    }
                );
            }
            // No exception is expected in this example, but if one is still thrown from a task,
            // it will be wrapped in AggregateException and propagated to the main thread.
            catch (AggregateException e)
            {
                Console.WriteLine("An action has thrown an exception. THIS WAS UNEXPECTED.\n{0}", e.InnerException.ToString());
            }
        }

        static void BasicAction()
        {
            Console.WriteLine("Method=alpha, Thread={0}", Thread.CurrentThread.ManagedThreadId);
        }

3.Parallel.For方法,主要用于处理针对数组元素的并行操作(数据的并行) 

       static void Main(string[] args)
        {
            int[] nums = new int[] { 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12 };
            Parallel.For(0, nums.Length, (i) =>
            {
                Console.WriteLine("针对数组索引{0}对应的那个元素{1}的一些工作代码……ThreadId={2}", i, nums[i], Thread.CurrentThread.ManagedThreadId);
            });
            Console.ReadKey();
        }

4.Parallel.ForEach方法,主要用于处理泛型集合元素的并行操作(数据的并行)

using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;

namespace AsyncDemo1
{
    class Program
    {
        static void Main(string[] args)
        {
            List<Student> students = new List<Student>();
            students.Add(new Student() { 
                StudentID=1,
                Name="张三",
                Age=18
            });
            students.Add(new Student()
            {
                StudentID = 2,
                Name = "李四",
                Age = 17
            });
            students.Add(new Student()
            {
                StudentID = 3,
                Name = "王五",
                Age = 19
            });
            Parallel.ForEach(students, (item)=>
            {
                Console.WriteLine($"学生编号:{item.StudentID},姓名:{item.Name},年龄:{item.Age}");
            });
            Console.ReadLine();
        }
    }

    public class Student
    {
        public int StudentID { get; set; }
        public string Name { get; set; }
        public int Age { get; set; }
    }
}

5.ForEach的独到之处就是可以将数据进行分区,每一个小区内实现串行计算,分区采用Partitioner.Create实现。

        static void Main(string[] args)
        {
            ConcurrentBag<int> bag = new ConcurrentBag<int>();
            var watch = Stopwatch.StartNew();
            watch.Start();
            Parallel.ForEach(Partitioner.Create(0, 3000000), tuple =>
            {
                for (int m = tuple.Item1; m < tuple.Item2; m++)
                {
                    bag.Add(m);
                }
            });
            Console.WriteLine("并行计算:集合有:{0},总共耗时:{1}", bag.Count, watch.ElapsedMilliseconds);
            GC.Collect();
            Console.ReadLine();
        }

6.ParallelOptions类

有时候我们的线程可能会跑遍所有的内核,为了提高其他应用程序的稳定性,就要限制参与的内核,正好ParallelOptions提供了MaxDegreeOfParallelism属性。

  static void Main(string[] args)
        {
            List<Student> students = new List<Student>();
            students.Add(new Student()
            {
                StudentID = 1,
                Name = "张三",
                Age = 18
            });
            students.Add(new Student()
            {
                StudentID = 2,
                Name = "李四",
                Age = 17
            });
            students.Add(new Student()
            {
                StudentID = 3,
                Name = "王五",
                Age = 19
            });
            Parallel.ForEach(students, new ParallelOptions() { MaxDegreeOfParallelism = 2 }, (item) =>
                {
                    Console.WriteLine($"学生编号:{item.StudentID},姓名:{item.Name},年龄:{item.Age}");
                });
            Console.ReadLine();
        }
 public class Student
    {
        public int ID { get; set; }
        public string Name { get; set; }
        public int Age { get; set; }
        public DateTime CreateTime { get; set; }
    }

    class Program
    {
        static void Main(string[] args)
        {
            var dic = LoadData();
            Stopwatch watch = new Stopwatch();
            watch.Start();
            var query2 = (from n in dic.Values.AsParallel()
                          where n.Age > 20 && n.Age < 25
                          select n).ToList();
            watch.Stop();
            Console.WriteLine("并行计算耗费时间:{0}", watch.ElapsedMilliseconds);

            Console.Read();
        }

        public static ConcurrentDictionary<int, Student> LoadData()
        {
            ConcurrentDictionary<int, Student> dic = new ConcurrentDictionary<int, Student>();
            ParallelOptions options = new ParallelOptions();
            //指定使用的硬件线程数为4
            options.MaxDegreeOfParallelism = 4;
            //预加载1500w条记录
            Parallel.For(0, 15000000, options, (i) =>
            {
                var single = new Student()
                {
                    ID = i,
                    Name = "hxc" + i,
                    Age = i % 151,
                    CreateTime = DateTime.Now.AddSeconds(i)
                };
                dic.TryAdd(i, single);
            });

            return dic;
        }
    }

7.中途退出并行循环

       在串行代码中我们break一下就搞定了,但是并行就不是这么简单了,不过没关系,在并行循环的委托参数中提供了一个ParallelLoopState,该实例提供了Break和Stop方法来帮我们实现。
  Break: 当然这个是通知并行计算尽快的退出循环,比如并行计算正在迭代100,那么break后程序还会迭代所有小于100的。
  Stop:这个就不一样了,比如正在迭代100突然遇到stop,那它啥也不管了,直接退出。

 static void Main(string[] args)
        {
            ConcurrentBag<int> bag = new ConcurrentBag<int>();
            Parallel.For(0, 20000000, (i, state) =>
            {
                if (bag.Count == 1000)
                {
                    //state.Break();
                    state.Stop();
                    return;
                }
                bag.Add(i);
            });
            Console.WriteLine("当前集合有{0}个元素。", bag.Count);
        }

8.并行中的异常处理

首先任务是并行计算的,处理过程中可能会产生n多的异常,那么如何来获取到这些异常呢?普通的Exception并不能获取到异常,然而为并行诞生的AggregateExcepation就可以获取到一组异常。

       static void Main(string[] args)
        {
            try
            {
                Parallel.Invoke(Run1, Run2);
            }
            catch (AggregateException ex)
            {
                foreach (var single in ex.InnerExceptions)
                {
                    Console.WriteLine(single.Message);
                }
            }
            Console.WriteLine("结束了!");
            //Console.Read();
        }

        static void Run1()
        {
            Thread.Sleep(3000);
            throw new Exception("我是任务1抛出的异常");
        }

        static void Run2()
        {
            Thread.Sleep(5000);
            throw new Exception("我是任务2抛出的异常");
        }

还有另外一种就是直接在Parallel处理异常

 private static readonly object locker = new object();
        static void Main(string[] args)
        {
            List<string> errList = new List<string>();
            Parallel.For(0, 10, (i) =>
            {
                try
                {
                    TestClass a = new TestClass();
                    a.Test(i);
                }
                catch (Exception ex)
                {
                    lock (locker)
                    {
                        errList.Add(ex.Message);
                    }
                    //Console.WriteLine(ex.Message);
                    //注:这里不再将错误抛出.....
                    //throw ex;
                }
            });

            int Index = 1;
            foreach (string err in errList)
            {
                Console.WriteLine("{0}、的错误:{1}", Index++, err);
            }
        }

参考链接:https://www.cnblogs.com/scmail81/p/9521096.html

               https://docs.microsoft.com/zh-cn/dotnet/api/system.threading.tasks.parallel?view=net-6.0

标签:Console,Age,---,static,WriteLine,new,多线程,Parallel
来源: https://www.cnblogs.com/hobelee/p/16514228.html