C#多线程（4）——任务并行库TPL

文章目录

- - 1 什么是TPL
  - 2 创建与启动任务
  - 3 等待任务
  - 4 任务中的异常处理
  - 5 取消任务

1 什么是TPL

$\textcolor{red}{TPL}$ （Task Parallel Library）任务并行库，是从.NetFramwork4.0后引入的基于异步操作的一组API。TPL的底层是基于多线程实现的，但是它相较于直接使用多线程，更为简单，它向程序员隐藏了与线程池交互的底层代码。在.NetFramwork4.0后，微软更推荐程序员使用TPL去编写多线程代码或者并行代码。
TPL的核心是任务，一个任务代表了一个异步操作，该操作可以使用或不适用独立的线程运行。
一个任务可以和其他的任务组合起来，比如同时启动多个任务，等待所有的任务完成；对之前所有任务的结果进行计算，TPL的优势在于具有组合任务API，而不用单独书写线程同步的代码（关注于锁、线程间的信号）。同样在多线程中关于多线程中异常的传播与处理是极为复杂的，而在TPL中，可以通过 $\textcolor{red}{AggregateException}$ ，捕获底层任务的所有异常，并允许单独处理这些异常。

2 创建与启动任务

创建、启动任务使用到的关键类如下：类定义在 $\textcolor{red}{System.Threading.Tasks}$ 命名空间中
在这里插入图片描述

       public static void Main(string[] args)
        {
            void TaskMethod(string name) {
                Console.WriteLine("Task{0} is runing on ThreadId {1} Is ThreadPool Thread {2}",
                    name,
                    Thread.CurrentThread.ManagedThreadId,
                    Thread.CurrentThread.IsThreadPoolThread);
            }


            int  TaskMethod_1(string name) {

                Console.WriteLine("Task{0} is runing on ThreadId {1} Is ThreadPool Thread {2}",
                     name,
                     Thread.CurrentThread.ManagedThreadId,
                     Thread.CurrentThread.IsThreadPoolThread);

                return 0;
            }


            #region 1 创建任务
            TaskMethod("Main");
            //public Task(Action action) 传入一个无返回的委托函数
            Task t1 = new Task(() => TaskMethod("t1"));
            Task t2 = new Task(() => TaskMethod("t2"));
            t1.Start();//显示创建Task需要运行start方法才能运行任务中的方法
            t2.RunSynchronously(); //RunSynchronously()在当前线程上运行任务方法
            Task.Run(() => TaskMethod("t3"));
            Task.Factory.StartNew(() => TaskMethod("t4"));


            //Task<T> 管理有返回值的工作单元  public Task(Func<TResult> function)
            //通过传入有返回的委托函数来创建任务对象
            Task<int> t5 = new Task<int>(() => TaskMethod_1("t5"));
            t5.Start();
            Console.WriteLine(t5.Result); //获得任务结果

            Console.ReadKey();
            #endregion

        }

在这里插入图片描述

3 等待任务

有以下方式可以等待任务完成：

调用Wait方法（可选择指定超时时间）
访问Result属性（当使用Task时）
Task.WaitAll（等待所有指定任务完成）
Task.WaitAny（等待任意一个任务完成）。

            #region 2 等待任务
         
            Console.OutputEncoding = Encoding.Unicode;
            Console.WriteLine("起始执行时间为{0}", DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss"));

            List<Task> list = new List<Task>();
            
            for (int i = 1; i <= 5; i++)
            {
                int tempI = i; //需要使用局部变量，因为for循环在数据量很小的时候，for循环结束时 task启动了，但是可能还未执行。由于共享变量i，所有在真正执行Task时，线程名称将一样 i=11
                Task t = new Task(()=>TaskMethod_2("-"+ tempI, tempI));
                list.Add(t);
                t.Start();//因为多线程的启动并不意味着立马进行，需要等待操作系统的调度。
            }

            Task.WaitAll(list.ToArray());//等待所有的线程完成
            Task.WaitAny(list.ToArray());//等待任意一个线程完成后执行，相当于在一个ManualResetEventSlim上等待，

            Console.WriteLine("等待所有任务线程执行完成,结束执行时间为{0}", DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss"));


            Task<int> t_1 = new Task<int>(() => TaskMethod_2("t_1", 10));
            t_1.Start();
            t_1.Wait();//主线程等待t_1线程完成任务方法。 Wait(TimeSpan timeout) 也可以等待具体的时间
            Console.WriteLine(t_1.Result);

            Console.WriteLine("结束执行时间为{0}", DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss"));
            Console.ReadKey();
            #endregion

TaskMethod_2 方法如示例一代码块。方法运行后如下

在这里插入图片描述

4 任务中的异常处理


            #region 3 任务的异常处理
            Task t1 = Task.Factory.StartNew(() => {
                throw new Exception("Task Failed ! ");
            });
            try
            {
                t1.Wait();//当你等待一个任务结束时（通过调用Wait方法或访问其Result属性），所有未处理的异常都会用一个AggregateException对象封装，方便重新抛给调用方。
            }
            catch (AggregateException aex)
            {
                Console.WriteLine(aex.InnerException.Message);  // Task Failed !
            }

            //定义一个父子任务，在父任务与子任务中分别抛出异常
            TaskCreationOptions atp = TaskCreationOptions.AttachedToParent;
            var parent = Task.Factory.StartNew(() =>
            {
                Console.WriteLine("I am Parent");

                Task.Factory.StartNew(() =>   // 子
                {
                    Console.WriteLine("I am Child");
                    throw new Exception("Child Exception");
                }, atp);

                throw new Exception("Parent Exception");

            });

            try {
                parent.Wait();
            }
            catch (AggregateException aex)
            {
                Console.WriteLine(aex.InnerExceptions.Count);  //2 捕获了父异常和子异常
                //这是对于有父子关系的任务，在父任务上等待也会隐式的等待子任务，所有子任务的
                //异常也会传递出来。
            }

            Console.ReadKey();
            #endregion

main 方法块如上，运行结果如下图所示
在这里插入图片描述

当某个任务抛出一个或多个异常时，异常包装在 $\textcolor{blue}{AggregateException }$ 异常中。该异常会传播回与任务联接的线程。通常，该线程是 $\textcolor{red}{等待任务完成}$ 的线程
（1）Wait方法
（2）WaitAny方法
（3）WaitAll方法
或访问 Result 属性的线程。AggregateException 通常包含关联任务线程中的所有异常，我们可以在外部通过try catch的方式去处理它，但是这并不意味着并不需要单独处理任务线程的异常，否则可能会因为无解的异常导致程式的中断。

5 取消任务

任务的取消需要用到 $\textcolor{red}{CancellationTokenSource}$ 和 $\textcolor{red}{CancellationToken}$ 类。在取消任务的过程中需要了解一下几点：

可以在构造中传入CancellationToken来构建Task任务，并且CancellationToken可以绑定到多个任务上
Task的创建和执行都是独立的，如果在任务执行前取消了任务，那么任务代码将不会执行。如果尝试调用start方法，将会抛出异常InvalidOperationException
任务执行后去执行CancellationTokenSource.Cancel方法，任务不会被取消
需要在任务代码中 $\textcolor{blue}{显示定义任务中断的逻辑}$

Main 方法代码块如下


            #region 5 取消任务
            
            Console.OutputEncoding = System.Text.Encoding.UTF8;
            void RunTask(string name) {
                for (int i = 0; i < 10; i++) {
                    Console.WriteLine(name+"运行开始:" + i);
                    Thread.Sleep(1000);
                }
            }

            void RunTaskWithCancellationToken(string name,CancellationToken token) {
                for (int i = 0; i < 10; i++)
                {
                    Console.WriteLine(name + "运行开始:" + i);
                    Thread.Sleep(1000);
                    token.ThrowIfCancellationRequested(); //在运行的Task任务中，显示定义任务取消的逻辑
                }
            }

            CancellationTokenSource source = new CancellationTokenSource();
            CancellationToken cancellation = source.Token;

            //创建一个和CancellationToken关联的任务类
            //t1任务不支持显示取消任务
            Task t1 = Task.Factory.StartNew(() => RunTask("t1"), cancellation);
            //t2任务支持显示取消任务，在任务执行的逻辑中加了ThrowIfCancellationRequested，标识希望在任务中断是抛出异常
            Task t2 = Task.Factory.StartNew(() => RunTaskWithCancellationToken("t2",cancellation), cancellation);
            //t3任务不支持显示取消任务，且需要手动start去运行任务代码。
            Task t3 = new Task(()=>RunTask("t1"),cancellation);
            
            Thread.Sleep(TimeSpan.FromSeconds(3));

            source.Cancel();

            try
            {
                //在一个任务调度前，取消任务，那么将会抛出System.InvalidOperationException 标识在已经完成的工作上呼叫start动作。
                t3.Start();
            }
            catch (Exception e)
            {

                Console.WriteLine("异常类型{0} ,异常消息{1}",e.GetType().Name,e.Message);
            }


            Console.WriteLine("t1 IsCanceled  {0} ,t1 IsCompleted{1}, t1 IsFaulted{2}, t1 status{3}", t1.IsCanceled, t1.IsCompleted, t1.IsFaulted, t1.Status.ToString());
            Console.WriteLine("t2 IsCanceled  {0} ,t2 IsCompleted{1}, t2 IsFaulted{2}, t2 status{3}", t2.IsCanceled, t2.IsCompleted, t2.IsFaulted, t2.Status.ToString());
            Console.WriteLine("t3 IsCanceled  {0} ,t3 IsCompleted{1}, t3 IsFaulted{2}, t3 status{3}", t3.IsCanceled, t3.IsCompleted, t3.IsFaulted, t3.Status.ToString());
            Console.ReadKey();
            #endregion