对话Task

Posted 2020-11-20 chengtian

上一篇简单讲解了 线程和线程池以及上下文切换。创建线程代价高昂，默认每个线程都要占用大量虚拟内存1M。更有效的做法使用线程池，重复利用线程。在.NET4.0中引入了TPL任务并行库，你可以在将精力集中于程序要完成的工作，同时最大程度地提高代码的性能。在C#5.0中引入了async 和 await关键字，基于任务的异步模式(TAP),所以了解Task对后面学习异步操作会简单些。 

任务是封装了以异步方式执行的工作。当启动一个任务，控制几乎立即返回调用者，无论任务要执行多少工作。

创建Task任务

 有三种创建方式

• 使用task构造函数
• task工厂类静态方法
• 使用.NET4.5新引入的Task.run()。

我们创建一个输出300万个32位字符的GUID任务分别使用三种不同方式实现。代码如下 constint RepeatCount = 1000000; //重复次数

            var listGuid = new BlockingCollection<string>();

            //Action无返回值 
            Action dowork = () =>
           {
               for (var count = 0; count < RepeatCount; count++)
               {
                   listGuid.Add(Guid.NewGuid().ToString("N"));
               }
           };
            Task task1 = new Task(dowork);  //1)使用构造函数
            task1.Start();
            Task task2 = Task.Factory.StartNew(dowork); //2)Task工厂方法,直接运行，不需要在调用start()
            Task task3 = Task.Run(dowork); //3)4.5 Task.Run 是Task.Factory.StartNew简化方式；直接运行，不需要在调用start()

            Task.WaitAll(task1, task2, task3); //等待所有任务完成，相当于 thread.join()
            Console.Write($"生成数量：{listGuid.Count / 10000}万");

输出

上述实例创建一个没有返回值的任务，当然也可以通过Task<TResult> 来创建返回值的异步操作。

连续任务

第一个任务生成32位字符的Guid任务，利用返回的结果再转化成对应的ASCII码，最后ASCII码十进制的值相加。代码如下 

 //Func 
            Func<string> doWork = () =>
            {
                return Guid.NewGuid().ToString("N");
            };
            //延续任务
            var task = Task.Run(doWork).ContinueWith(async strGuid =>
            {
                var resut = await strGuid;
                var array = Encoding.ASCII.GetBytes(resut);

                int mLenght = array.Length;
                int sumResult = 0;
                for (int m = 0; m < mLenght; m++)
                {
                    sumResult += array[m];
                }
                Console.WriteLine($"Guid对应10进制相加结果:{sumResult}");
            });

输出 

处理任务异常

同步代码要想捕获异常，只需在代码块上添加Try ...Catch即可。但是异步调用不能这么做。因为控制会立即从调用返回，然后控制会离开Try块，而这时距离工作者线程发生异常可能还有好久呢。

为了处理出错的任务，一个技术是显式创建延续任务作为那个任务的“错误处理程序”。检测到先驱任务引发未处理的异常，任务调度器会自动调度延续任务。但是，如果没有这种处理程序，同时在出错的任务上执行wait()（或其他试图获取result的动作），就会引发一个AggregateException，示例代码如下。

  Task task = Task.Run(() =>
            {
                throw new InvalidOperationException();
            });

            try
            {
                task.Wait();
            }
            catch (Exception ex)
            {
                Console.WriteLine($"常规erro:{ex.Message};type:{ex.GetType()}");
                AggregateException excetion = (AggregateException)ex;
                excetion.Handle(eachException =>
                {
                    Console.WriteLine($"erro:{eachException.Message}");
                    return true;
                });
            }

输出

虽然工作者线程上已发的未处理异常是InvalidOperationException类型，但主线程捕捉的仍是一个AggregateException。由于编译时不知道工作者任务将要引发一个还是多个异常，所以未处理的出错任务总是引发一个AggregateException。

  还可查看任务的Exception属性来了解出错任务的状态，这样不会造成在当前线程上重新引发异常。代码如下

  bool paraentTaskFaulted = false;
            Task task = new Task(() =>
            {
                throw new InvalidOperationException();
            });

            Task continuationTask = task.ContinueWith(t =>
            {

                paraentTaskFaulted = t.IsFaulted;
            }, TaskContinuationOptions.OnlyOnFaulted);

            task.Start();
            Console.Write(continuationTask.Status);
            continuationTask.Wait();
            //如果断言失败 则显示一个消息框，其中显示调用堆栈。
            Trace.Assert(paraentTaskFaulted);
            if (!task.IsFaulted)
            {
                task.Wait();
            }
            else
            {
                task.Exception.Handle(eachException =>
                {
                    Console.WriteLine($"erro:{eachException.Message}");
                    return true;
                });
            }

 

注意，为了获取原始任务上的未处理异常，我们使用Exception属性。结果和上面示例输出一样。

取消任务

 任务支持取消，比如常用在指定时间内的任务或者基于某些条件手动的取消，支持取消的任务要监听一个CancellationToken对象。任务轮询它，检查是否出发了取消请求。如下代码展示了取消请求和对请求的响应。

 /// <summary>
        /// 取消任务
        /// </summary>
        public void TaskTopic5()
        {
            string stars = "*".PadRight(Console.LargestWindowWidth-1,‘*‘);
            Console.WriteLine("push enter to exit.");
            CancellationTokenSource cancellationTokenSource = new CancellationTokenSource(); //向应该被取消的 System.Threading.CancellationToken 发送信号
            Task task = Task.Run(
                ()=>
                Count(cancellationTokenSource.Token,100),
                cancellationTokenSource.Token);

            Console.Read();
            cancellationTokenSource.Cancel();//按下enter键， 传达取消请求
            
            Console.WriteLine(stars);
            Console.WriteLine(task.IsCanceled);
            task.Wait();
            Console.WriteLine();
        }
        /// <summary>
        /// 数数
        /// </summary>
        /// <param name="token"></param>
        /// <param name="countTo"></param>
        private void Count(CancellationToken token,int countTo)
        {
            for (int count = 1; count < countTo; count++)
            {
                //监控是否取消
                if (token.IsCancellationRequested)
                {
                    Console.WriteLine("数数喊停了");
                    break;
                }

                Console.Write(count+"=》");
                Thread.Sleep(TimeSpan.FromSeconds(1));
            }
            Console.WriteLine("数数结束");
        }

 

输出

  调用Cancel()实际会在从cancellationTokenSource.Token复制的所有取消标志上设置IsCancellationRequested属性。

 到此任务的一些基本的操作已经完成了，下一节关注下C#5.0的async/await上下文关键字。