C#中的thread和task之Task

简介

Task是在.NET Framework 4中添加进来的，这是新的namespace：System.Threading.Tasks;它强调的是adding parallelism and concurrency to applications。

现在都是多核的CPU，在系统内，Task Parallel Library能更有效地利用CPU核心。TPL 会动态地按比例调节并发程度，以便最有效地使用所有可用的处理器。TPL 还处理工作分区、ThreadPool 上的线程调度、取消支持、状态管理以及其他低级别的细节操作。对比ThreadPool上的Thread并没有很好地支持Cancel的操作。

在语法上，和lambda表达式更好地结合。

创建Task

显式创建Task

先看一个简单的示例：

using System; using System.Threading; using System.Threading.Tasks; public class TaskExample1 { public static void Main() { Thread.CurrentThread.Name = "Main"; // 使用lambda方式，之间提供一个用户delegate，创建一个Task Task taskA = new Task( () => Console.WriteLine("From taskA.")); // Start taskA.Start(); // Output a message from the calling thread. Console.WriteLine("From thread '{0}'.", Thread.CurrentThread.Name); //等待task结束 taskA.Wait(); } } // From thread 'Main'. // From taskA.

任务对象Task提供可在任务的整个生存期内从调用线程访问的方法和属性。例如，可以随时访问任务的 Status 属性，以确定它是已开始运行、已完成运行、已取消还是引发了异常。状态由 TaskStatus 枚举表示。

简化创建并开始的Task操作版本–直接使用一个操作： System.Threading.Tasks.Task.Run()

using System; using System.Threading; using System.Threading.Tasks; public class TaskExample2 { public static void Main() { Thread.CurrentThread.Name = "Main"; // Define and run the task. Task taskA = Task.Run( () => Console.WriteLine("From taskA.")); // Output a message from the calling thread. Console.WriteLine("From thread '{0}'.", Thread.CurrentThread.Name); taskA.Wait(); } } // From thread 'Main'. // From taskA.

TaskFactory创建Task

TPL提供了工厂类TaskFactory，也可用直接创建Task。 一个操作System.Threading.Tasks.TaskFactory.StartNew， 即可完成创建并开始一个Task.

Use this method when creation and scheduling do not have to be separated and you require additional task creaion options or the use of a specific scheduler, or when you need to pass additional state into the task through its AsyncState property, as shown in the following example.

using System; using System.Threading; using System.Threading.Tasks; public class TaskExample3 { public static void Main() { Thread.CurrentThread.Name = "Main"; // 创建task并启动 Task taskA = Task.Factory.StartNew(() => Console.WriteLine("From taskA.")); // Console.WriteLine("From thread '{0}'.", Thread.CurrentThread.Name); // wait task taskA.Wait(); } } // Hello from thread 'Main'. // Hello from taskA.

传入参数

来自MS的示例：

using System; using System.Threading; using System.Threading.Tasks; public class Example { public static void Main() { Task[] taskArray = new Task[10]; for (int i = 0; i < taskArray.Length; i++) { taskArray[i] = Task.Factory.StartNew( (int param) => { Console.WriteLine("Task #{0}.", param); }, i ); } Task.WaitAll(taskArray); } }

Task返回值

如果一个Task执行方法是有返回值的，可用得到其值。在创建一个task时，其返回的值为Task，表示一个返回类型为T的Task；在taks完成执行后，可用通过Task的Result属性获取结果。

public class Task<TResult> : System.Threading.Tasks.Task

来自MS的示例：

using System; using System.Linq; using System.Threading.Tasks; class Program { static void Main() { // lambda表达式创建一个task并执行，并返回值。 Task<int> task1 = Task<int>.Factory.StartNew(() => 1); int i = task1.Result; // 返回对象 Task<Test> task2 = Task<Test>.Factory.StartNew(() => { string s = ".NET"; double d = 4.0; return new Test { Name = s, Number = d }; }); Test test = task2.Result; // Return an array produced by a PLINQ query Task<string[]> task3 = Task<string[]>.Factory.StartNew(() => { string path = @"C:\Users\Public\Pictures\Sample Pictures\"; string[] files = System.IO.Directory.GetFiles(path); var result = (from file in files.AsParallel() let info = new System.IO.FileInfo(file) where info.Extension == ".jpg" select file).ToArray(); return result; }); foreach (var name in task3.Result) Console.WriteLine(name); } class Test { public string Name { get; set; } public double Number { get; set; } } }

Continue操作

如果一个Task开始执行时间依赖于其它的Task的完成，可以使用Continue系列方法。

来自MS的示例：

using System; using System.Threading.Tasks; public class Example { public static void Main() { var getData = Task.Factory.StartNew(() => { Random rnd = new Random(); int[] values = new int[100]; for (int ctr = 0; ctr <= values.GetUpperBound(0); ctr++) values[ctr] = rnd.Next(); return values; } ); var processData = getData.ContinueWith((x) => { int n = x.Result.Length; long sum = 0; double mean; for (int ctr = 0; ctr <= x.Result.GetUpperBound(0); ctr++) sum += x.Result[ctr]; mean = sum / (double) n; return Tuple.Create(n, sum, mean); } ); var displayData = processData.ContinueWith((x) => { return String.Format("N={0:N0}, Total = {1:N0}, Mean = {2:N2}", x.Result.Item1, x.Result.Item2, x.Result.Item3); } ); Console.WriteLine(displayData.Result); } } // 输出： // N=100, Total = 110,081,653,682, Mean = 1,100,816,536.82

而且，可以使用更简化的写法.

using System; using System.Threading.Tasks; public class Example { public static void Main() { var displayData = Task.Factory.StartNew(() => { Random rnd = new Random(); int[] values = new int[100]; for (int ctr = 0; ctr <= values.GetUpperBound(0); ctr++) values[ctr] = rnd.Next(); return values; } ). ContinueWith((x) => { int n = x.Result.Length; long sum = 0; double mean; for (int ctr = 0; ctr <= x.Result.GetUpperBound(0); ctr++) sum += x.Result[ctr]; mean = sum / (double) n; return Tuple.Create(n, sum, mean); } ). ContinueWith((x) => { return String.Format("N={0:N0}, Total = {1:N0}, Mean = {2:N2}", x.Result.Item1, x.Result.Item2, x.Result.Item3); } ); Console.WriteLine(displayData.Result); } }

Cancel操作

在TPL中，为Task能cancel执行，提供了CancellationTokenSource和CancellationToken； 需要完成的工作是：在Task的action执行方法内，周期性地检查CancellationToken的IsCancellationRequested属性。 示例：

private var tokenSource = new CancellationTokenSource(); public void Start() { var token = tokenSource.Token; for (int i = 0; i<5; i++>) { Task t = Task.Factory.StartNew( () => DoSomeWork(i, token), token); Console.WriteLine("Task {0} executing", t.Id); } } public void Stop() { var token = tokenSource.Token; tokenSource.Cancel(); } void DoSomeWork(int taskNum, CancellationToken ct) { // 先检查，调度进入时，是否cancel了。 if (ct.IsCancellationRequested == true) { Console.WriteLine("Task {0} was cancelled before it got started.", taskNum); ct.ThrowIfCancellationRequested(); // 抛出异常-- 或者 return } // 正式开始任务。 int maxIterations = 100; // NOTE!!! A "TaskCanceledException was unhandled for (int i = 0; i <= maxIterations; i++) { // var sw = new SpinWait(); for (int j = 0; j <= 100; j++) sw.SpinOnce(); if (ct.IsCancellationRequested) { Console.WriteLine("Task {0} cancelled", taskNum); ct.ThrowIfCancellationRequested(); //抛出异常-- 或者 return } } }