任务并行库

带着问题去思考！大家好！

简介

之前我们说了线程池和线程以及运用。实际上可以理解为他只是一个抽象层，其向程序员隐藏了使用线程的细节。但是使用线程池也是相当复杂，接着我们运用异步编程模型和基于事件的异步模式，这样获取结果很容易，传播异常也很轻松，但是组合多个异步操作仍然需要大量的工作，所以在.NET Framework 4.5引入了一个新的异步操作的API，任务并行库（Task）.随之下一个版本进行更新。

APM API转换为任务

首先我们要知道什么是APM模式。

.net 1.0时期就提出的一种异步模式，并且基于IAsyncResult接口实现BeginXXX和EndXXX类似的方法。

 //APM API模式转换为任务.Net Core不支持
        private delegate string AsynchronousTask(string threadName);
        private delegate string IncompatibleAsynchronousTask(out int threadId);

        /// <summary>
        /// 回调
        /// </summary>
        /// <param name="ar"></param>
        private static void Callback(IAsyncResult ar)
        {
            Console.WriteLine("开始一个回调...");
            Console.WriteLine("传递给回调的状态：{0} ", ar.AsyncState);
            Console.WriteLine("是否是线程池线程：{0}", Thread.CurrentThread.IsThreadPoolThread);
            Console.WriteLine("线程池任务的线程id：{0}", Thread.CurrentThread.ManagedThreadId);
        }
        private static string Test(string threadName)
        {
            Console.WriteLine("开始.....");
            Console.WriteLine("是否是线程池线程:{0}", Thread.CurrentThread.IsThreadPoolThread);
            Thread.Sleep(TimeSpan.FromSeconds(2));
            Thread.CurrentThread.Name = threadName;
            return string.Format("线程名称:{0}", Thread.CurrentThread.Name);
        }
        private static string Test(out int threadId)
        {
            Console.WriteLine("开始.....");
            Console.WriteLine("是否是线程池线程:{0}", Thread.CurrentThread.IsThreadPoolThread);
            Thread.Sleep(TimeSpan.FromSeconds(2));
            threadId = Thread.CurrentThread.ManagedThreadId;
            return string.Format("线程池任务线程id是:{0}", threadId);
        }
        static void Main(string[] args)
        {
            int threadId;
            //委托赋值
            AsynchronousTask d = Test;
            IncompatibleAsynchronousTask e = Test;
            Console.WriteLine("选择1");
            Task<string> task = Task<string>.Factory.FromAsync(d.BeginInvoke("AsyncTaskThread", Callback, "a delegate asynchronous call"), d.EndInvoke);

            task.ContinueWith(t => Console.WriteLine("回调已经完成, now  running a continuation! Result:{0}", t.Result));

            while (!task.IsCompleted)
            {
                Console.WriteLine(task.Status);
                Thread.Sleep(TimeSpan.FromSeconds(0.5));
            }
            Console.WriteLine(task.Status);
            Thread.Sleep(TimeSpan.FromSeconds(1));


            Console.WriteLine("----------------------");
            Console.WriteLine("选择2");
            task = Task<string>.Factory.FromAsync(d.BeginInvoke, d.EndInvoke, "AsyncTaskThread", "a delegate asynchronous call");
            task.ContinueWith(t => Console.WriteLine("任务已经完成, now  running a continuation! Result:{0}", t.Result));
            while (!task.IsCompleted)
            {
                Console.WriteLine(task.Status);
                Thread.Sleep(TimeSpan.FromSeconds(0.5));
            }
            Console.WriteLine(task.Status);
            Thread.Sleep(TimeSpan.FromSeconds(1));


            Console.WriteLine("----------------------");
            Console.WriteLine("选择3");

            IAsyncResult ar = e.BeginInvoke(out threadId, Callback, "a delegate asynchronous call");
            ar = e.BeginInvoke(out threadId, Callback, "a delegate asynchronous call");
            task = Task<string>.Factory.FromAsync(ar, _ => e.EndInvoke(out threadId, ar));
            task.ContinueWith(t => Console.WriteLine("任务已经完成, now  running a continuation! Result:{0}", t.Result));
            while (!task.IsCompleted)
            {
                Console.WriteLine(task.Status);
                Thread.Sleep(TimeSpan.FromSeconds(0.5));
            }
            Console.WriteLine(task.Status);
            Thread.Sleep(TimeSpan.FromSeconds(1));

        }

这里我们定义两个委托，其中一个使用了OUT参数，因此再将APM模式转换为任务时，与标准的TPL API是不兼容的。

将APM转换为TPL的关键点是Task<T>.Factory.FromAsync方法，T是异步操作结果是类型。该方法有数个重载。在第一例子中传入了IAsyncResult和Func<IAsyncResult,string>,这是一个将IAsyncResult的实现参数并返回一个字符串的方法。由于第一个委托提供的EndMethod与该签名是兼容的，所以将该委托的异步调用转换为任务是可以的。

第二个例子与第一个非常相似，使用了不同的FromAsync方法重载，该重载并不允许指定一个将会在异步委托调用完成后被调用的回调函数，但我们可以使用后续操作替代它，但如果回调函数很重要，可以使用第一个例子的方法。

最后一个例子展示了一个小技巧，这次IncompatibleAsynchronousTask委托的EndMethod使用了out参数。与FromAsync方法并不兼容。然而，可以很容易的将EndMethod调用封装到一个lambda表达式中，从而适合任务工厂方法。

第一个任务状态为：WaitingForActivation，这意味TPL基础设施实际上还为启动任务

 

 实现取消

基于任务的异步操作实现取消流程。我们将学习如果正确的使用取消标志，以及在任务真正运行前如何得知其他是否被取消。

  /// <summary>
        /// 任务取消
        /// </summary>
        /// <param name="name"></param>
        /// <param name="seconds"></param>
        /// <param name="token"></param>
        /// <returns></returns>
        private static int TestMethod(string name,int seconds,CancellationToken token)
        {
            Console.WriteLine("task {0} is running on a thread id {1},Is thread pool thread {2}",name,Thread.CurrentThread.ManagedThreadId,
                Thread.CurrentThread.IsThreadPoolThread);
            for (int i = 0; i < seconds; i++)
            {
                Thread.Sleep(TimeSpan.FromSeconds(1));
                if (token.IsCancellationRequested)
                    return -1;
            }
            return 42 * seconds;
        }
        static void Main(string[] args)
        {
            var cts =new CancellationTokenSource();
            var longTask = new Task<int>(()=>TestMethod("task 1",10,cts.Token),cts.Token);
            Console.WriteLine(longTask.Status);
            cts.Cancel();
            Console.WriteLine(longTask.Status);
            Console.WriteLine("First task has been cancelled before execution");

            cts = new CancellationTokenSource();
            longTask = new Task<int>(() => TestMethod("task 2", 10, cts.Token), cts.Token);
            longTask.Start();
            for (int i = 0; i < 5; i++)
            {
                Thread.Sleep(TimeSpan.FromSeconds(0.5));
                Console.WriteLine(longTask.Status);
            }
            cts.Cancel();
            for (int i = 0; i < 5; i++)
            {
                Thread.Sleep(TimeSpan.FromSeconds(0.5));
                Console.WriteLine(longTask.Status);
            }
            Console.WriteLine("A task has been completed with result {0}.",longTask.Result); 
        }

 这里我们所以说的为TPL任务实现取消选择的例子

首先仔细看看longTask的创建代码。我们将底层任务传递一次取消标志，然后给任务构造函数再传递一次，为什么需要提供取消标志两次呢？

是这样的；如果在任务实际启动前取消它，该任务的TPL基础设施有责任处理该取消操作。因为这些代码根本不会执行，通过得到的第一个任务的状态可以知道它被取消了。如果尝试堆该任务调用Start方法，将会得到InvalidOperationException异常。

然后需要自己写代码来处理取消过程，这意味着我们对取消过程全权负责，并且在取消任务后，任务的状态仍然是RanToCompletion，因为从TPL的视角来看，该任务正常完成了它的工作。辨别这两种情况是很重要的，并且需要理解每种情况下职责的不同。

处理任务中的异常

抛出异常的不同情况

/// <summary>
        /// 任务异常
        /// </summary>
        /// <param name="name"></param>
        /// <param name="seconds"></param>
        /// <returns></returns>
        public static int TaskMethod(string name,int seconds)
        {
            Console.WriteLine("task {0} is running on a thread id {1},Is thread pool thread {2}", name, Thread.CurrentThread.ManagedThreadId,
              Thread.CurrentThread.IsThreadPoolThread);
            Thread.Sleep(TimeSpan.FromSeconds(seconds));
            throw new Exception("Boom");
            return 42 * seconds;
        }
        static void Main(string[] args)
        {
            Task<int> task;
            try
            {
                task =Task.Run(() => TaskMethod("Task 1", 2));
                int result = task.Result;
                Console.WriteLine("Result:{0}",result);
            }
            catch (Exception ex)
            {
                Console.WriteLine("Exception caught:{0}", ex);
            }
            Console.WriteLine("---------------------");
            Console.WriteLine();

            try
            {
                task = Task.Run(() => TaskMethod("Task 2", 2));
                int result = task.GetAwaiter().GetResult();
                Console.WriteLine("Result:{0}", result);

            }
            catch (Exception ex)
            {
                Console.WriteLine("Exception caught:{0}", ex);
            }
            Console.WriteLine("---------------------");
            Console.WriteLine();

            var t1 = new Task<int>(()=>TaskMethod("Task 3",3));
            var t2 = new Task<int>(()=>TaskMethod("Task 4",2));

            var complexTask = Task.WhenAll(t1, t2);
            var exceptionHandler = complexTask.ContinueWith(t => Console.WriteLine("Exception caught:{0}", t.Exception, TaskContinuationOptions.OnlyOnFaulted));
            t1.Start();
            t2.Start();

            Thread.Sleep(TimeSpan.FromSeconds(5));
        }

 

 

 当程序启动时，创建一个任务并尝试同步获取任务结果。Result属性的Get部分会使当前线程等待直到该任务完成，并将异常传播给当前线程。这种情况下，通过catch代码块可以很容易的捕获异常，但是异常是一个被封装的异常。叫做AggregateException。在这个例子中，它里面包含一个异常，因为只有一个任务抛出异常。可以访问InnerException属性来得到底层异常。

第二个例子与第一个例子相似，不同的是使用了GetAwaiter和GetResult方法来访问任务结果。这种情况下，无需封装异常，因为TPL基础设施会提取该异常。如果只有一个底层任务，那么一次只能获取一个原始异常，这种设计非常合适

最后一个例子展示了两个任务抛出异常的情形，现在使用后续操作来处理异常。只有之前的任务完成前有异常时，该后续操作才会被执行。通过给后续操作传递TaskContinuationOptions.OnlyOnFaulted选项可以实现该行为。