C#之使用任务并行库

.NET Framework4.0引入了一个新的关于异步操作的API，它叫做任务并行库（Task Parallel Library，简称TPL)。TPL的核心是任务，一个任务代表一个异步操作，该操作可以通过多种方式运行，可以使用或不使用独立线程运行。

一个任务可以它通过多种方式与其他方式组合起来，TPL与之前的模式相比，其中一个关键优势就在于其具有用于组合任务的便利的API。

处理任务中的异常结果有多种方式。由于一个任务可能由其他任务组成，这些任务也可能拥有各自的子任务，所以有一个AggregateException的概念。这种异常可以捕获底层任务内部的所有异常，并允许单独处理这些异常。

创建任务

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

namespace ThreadDemo
{
    class Proram8
    {
        static void TaskMethod(string name)
        {
            Console.WriteLine($"Task {name} is running on a thread id" +
                $"{Thread.CurrentThread.ManagedThreadId}. Is thread pool thread:" +
                $"{Thread.CurrentThread.IsThreadPoolThread}");
        }
        static void Main()
        {
            var t1 = new Task(() => TaskMethod("Task 1"));
            var t2 = new Task(() => TaskMethod("Task 2"));
            t2.Start();
            t1.Start();
            Task.Run(() => TaskMethod("Task 3"));
            Task.Factory.StartNew(() => TaskMethod("Task4"));
            Task.Factory.StartNew(() => TaskMethod("Task5"),TaskCreationOptions.LongRunning);
            Thread.Sleep(TimeSpan.FromSeconds(1));
            
        }
    }
}

output:

Task Task 3 is running on a thread id5. Is thread pool thread:True
Task Task 2 is running on a thread id3. Is thread pool thread:True
Task Task4 is running on a thread id7. Is thread pool thread:True
Task Task5 is running on a thread id8. Is thread pool thread:False
Task Task 1 is running on a thread id4. Is thread pool thread:True

创建任务主要有两种方式，（1）用其构造函数，然后调用Start()方法（2)用Task的静态方法：Run(),Factory.StartNew()，静态方法会立即开始工作，而Run()只是Factory.StartNew()的一个快捷方式，后者有附加选项，如果我们对此附加选项标记为长时间运行，则该任务将不会使用线程池

使用任务执行基本的操作

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.Threading;
namespace ThreadDemo
{
    class Program9
    {
        static int TaskMethod(string name)
        {
            Console.WriteLine($"Task {name} is running on a thread id" +
                $"{Thread.CurrentThread.ManagedThreadId}. Is thread pool thread:" +
                $"{Thread.CurrentThread.IsThreadPoolThread}");
            Thread.Sleep(TimeSpan.FromSeconds(2));
            return 42;
        }
        static Task CreateTask(string name)
        {
            return new Task(() => TaskMethod(name));
        }
        static void Main()
        {
            TaskMethod("Main Thread Task");//并没有封装到任务中，就是主线程普通的函数
            var task = CreateTask("Task 1");
            task.Start();
            var result = task.Result;//该行为会阻塞主线程
            Console.WriteLine($"Result:{result}");

            task = CreateTask("Task 2");
            task.RunSynchronously();//该任务会运行在主线程上，可以避免使用线程池来执行非常短暂的操作
            result = task.Result;
            Console.WriteLine($"Resultt:{result}");

            task = CreateTask("Task 3");
            Console.WriteLine(task.Status);//打印Start之前的状态
            task.Start();
            while (task.IsCompleted)
            {
                Console.WriteLine(task.Status);
                Thread.Sleep(TimeSpan.FromSeconds(0.5));
            }
            Console.WriteLine(task.Status);
            result = task.Result;//阻塞主线程
            Console.WriteLine($"Result is {result}");

        }
    }
}

output:

Task Main Thread Task is running on a thread id1. Is thread pool thread:False
Task Task 1 is running on a thread id3. Is thread pool thread:True
Result:42
Task Task 2 is running on a thread id1. Is thread pool thread:False
Resultt:42
Created
Task Task 3 is running on a thread id4. Is thread pool thread:True
Running
Result is 42

组合任务

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.Threading;
namespace ThreadDemo
{
    class Program10
    {
        static int TaskMethod(string name,int seconds)
        {
            Console.WriteLine($"Task {name} is running on a thread id" +
                $"{Thread.CurrentThread.ManagedThreadId}.Is thread pool thread:" +
                $"{Thread.CurrentThread.IsThreadPoolThread}");
            Thread.Sleep(TimeSpan.FromSeconds(seconds));
            return 42 * seconds;
        }
        static void Main()
        {
            var firstTask = new Task(() => TaskMethod("First Task", 3));
            var secondTask = new Task(() => TaskMethod("Second Task", 2));
            firstTask.ContinueWith(
                t => Console.WriteLine($"The first answer is{t.Result}" +
                $"{Thread.CurrentThread.ManagedThreadId},is thread pool thread:" +
                $"{Thread.CurrentThread.IsThreadPoolThread}"),
                TaskContinuationOptions.OnlyOnRanToCompletion
                ) ;
            firstTask.Start();
            secondTask.Start();
            Thread.Sleep(TimeSpan.FromSeconds(4));

            Task continuation = secondTask.ContinueWith(t=>Console.WriteLine($"The second answer is" +
                $"{t.Result}. Thread id" +
                $"{Thread.CurrentThread.ManagedThreadId},is thread pool thread:" +
                $"{Thread.CurrentThread.IsThreadPoolThread}"),
                TaskContinuationOptions.OnlyOnRanToCompletion|TaskContinuationOptions.ExecuteSynchronously
                );
            continuation.GetAwaiter().OnCompleted(()=>Console.WriteLine($"" +
                $"Continuation Task completed! Thread id" +
                $"{Thread.CurrentThread.ManagedThreadId},is thread pool thread:" +
                $"{Thread.CurrentThread.IsThreadPoolThread}"));
            Thread.Sleep(TimeSpan.FromSeconds(2));
            Console.WriteLine();
            firstTask = new Task(()=>
            {
                var innerTask = Task.Factory.StartNew(()=>TaskMethod("Second Task",5),TaskCreationOptions.AttachedToParent);
                innerTask.ContinueWith(t => TaskMethod("Third Task", 2), TaskContinuationOptions.AttachedToParent);
                return TaskMethod("First Task", 2);
            });
            firstTask.Start();
            while (!firstTask.IsCompleted)
            {
                Console.WriteLine(firstTask.Status);
                Thread.Sleep(TimeSpan.FromSeconds(0.5));
            }
            Console.WriteLine(firstTask.Status);
            Thread.Sleep(TimeSpan.FromSeconds(10));
        }
    }
}

output:

Task First Task is running on a thread id3.Is thread pool thread:True
Task Second Task is running on a thread id4.Is thread pool thread:True
The first answer is1263,is thread pool thread:True
The second answer is84. Thread id1,is thread pool thread:False
Continuation Task completed! Thread id4,is thread pool thread:True

Running
Task Second Task is running on a thread id5.Is thread pool thread:True
Task First Task is running on a thread id6.Is thread pool thread:True
Running
Running
Running
WaitingForChildrenToComplete
WaitingForChildrenToComplete
WaitingForChildrenToComplete
WaitingForChildrenToComplete
WaitingForChildrenToComplete
WaitingForChildrenToComplete
Task Third Task is running on a thread id4.Is thread pool thread:True
WaitingForChildrenToComplete
WaitingForChildrenToComplete
WaitingForChildrenToComplete
WaitingForChildrenToComplete
RanToCompletion

将APM模式转换为任务

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.Threading;
namespace ThreadDemo
{
    class Program11
    {
        delegate string AsynchronousTask(string threadName);
        delegate string IncompatibleAsynchronousTask(out int threadId);
        static void Callback(IAsyncResult ar)
        {
            Console.WriteLine("Startting a callback...");
            Console.WriteLine($"State passed to a callback:{ar.AsyncState}");
            Console.WriteLine($"Is thread pool thread:" +
                $"{Thread.CurrentThread.IsThreadPoolThread}");
            Console.WriteLine($"Thread pool worker thread id:{Thread.CurrentThread.ManagedThreadId}");
        }
        static string Test(string threadName)
        {
            Console.WriteLine("Starting ....");
            Console.WriteLine($"Is thread pool thread:" +
                $"{Thread.CurrentThread.IsThreadPoolThread}");
            Thread.Sleep(TimeSpan.FromSeconds(2));
            Thread.CurrentThread.Name = threadName;
            return $"Thread name:{Thread.CurrentThread.Name}";
        }
        static string Test(out int threadId)
        {
            Console.WriteLine("Startting...");
            Console.WriteLine($"Is thread poool :{Thread.CurrentThread.IsThreadPoolThread}");
            Thread.Sleep(TimeSpan.FromSeconds(2));
            threadId = Thread.CurrentThread.ManagedThreadId;
            return $"Thread pool worker thread id:{threadId}";
        }
        public static void Main()
        {
            int threadId;
            AsynchronousTask d = Test;
            IncompatibleAsynchronousTask e = Test;

            Console.WriteLine("Option 1:");
            Task task = Task.Factory.FromAsync(
                d.BeginInvoke("AsyncTaskThread", Callback, "a delegate asynchronous call"), d.EndInvoke);
            task.ContinueWith(t => Console.WriteLine($"" +
                $"Callback is finished,now running a continuation!Result:{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();
            Console.WriteLine("Option 2:");
            task = Task.Factory.FromAsync(d.BeginInvoke, d.EndInvoke, "AsyncTaskThread", "a delegate asynchronous call");
            task.ContinueWith(t => Console.WriteLine($"Task is completed,now running a continuation!Resultt:{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();
            Console.WriteLine("Option 3:");
            IAsyncResult ar = e.BeginInvoke(out threadId, Callback, "a delegate asynchronous call");
            task = Task.Factory.FromAsync(ar, _ => e.EndInvoke(out threadId, ar));
            task.ContinueWith(t =>
            Console.WriteLine($"Task is completed,now running a continuation!" +
            $"Result:{t.Result},threadId:{threadId}"));
            while (!task.IsCompleted)
            {
                Console.WriteLine(task.Status);
                Thread.Sleep(TimeSpan.FromSeconds(0.5));
            }
            Console.WriteLine(task.Status);
            Thread.Sleep(TimeSpan.FromSeconds(1));
        }
    }
}

output:

Option 1:
Starting ....
Is thread pool thread:True
WaitingForActivation
WaitingForActivation
WaitingForActivation
WaitingForActivation
Startting a callback...
State passed to a callback:a delegate asynchronous call
Is thread pool thread:True
Thread pool worker thread id:3
Callback is finished,now running a continuation!Result:Thread name:AsyncTaskThread
RanToCompletion
----------------------------------------------------

Option 2:
WaitingForActivation
Starting ....
Is thread pool thread:True
WaitingForActivation
WaitingForActivation
WaitingForActivation
Task is completed,now running a continuation!Resultt:Thread name:AsyncTaskThread
RanToCompletion
-------------------------------------------------------

Option 3:
WaitingForActivation
Startting...
Is thread poool :True
WaitingForActivation
WaitingForActivation
WaitingForActivation
Startting a callback...
State passed to a callback:a delegate asynchronous call
Is thread pool thread:True
Thread pool worker thread id:3
Task is completed,now running a continuation!Result:Thread pool worker thread id:3,threadId:3
RanToCompletion

未完待续...