C#教程之C#多线程编程系列（五）- 使用任务并行

1.1 简介#

在之前的几个章节中，就线程的使用和多线程相关的内容进行了介绍。因为线程涉及到异步、同步、异常传递等问题，所以在项目中使用多线程的代价是比较高昂的，需要编写大量的代码来达到正确性和健壮性。

为了解决这样一些的问题，在.Net Framework 4.0中引入了一个关于一步操作的API。它叫做任务并行库(Task Parallel Library)。然后在.Net Framwork 4.5中对它进行了轻微的改进，本文的案例都是用最新版本的TPL库，而且我们还可以使用C# 5.0的新特性await/async来简化TAP编程，当然这是之后才介绍的。

TPL内部使用了线程池，但是效率更高。在把线程归还回线程池之前，它会在同一线程中顺序执行多少Task，这样避免了一些小任务上下文切换浪费时间片的问题。

任务是对象，其中封装了以异步方式执行的工作，但是委托也是封装了代码的对象。任务和委托的区别在于，委托是同步的，而任务是异步的。

在本章中，我们将会讨论如何使用TPL库来进行任务之间的组合同步，如何将遗留的APM和EAP模式转换为TPL模式等等。

1.2 创建任务#

在本节中，主要是演示了如何创建一个任务。其主要用到了System.Threading.Tasks命名空间下的Task类。该类可以被实例化并且提供了一组静态方法，可以方便快捷的创建任务。

在下面实例代码中，分别延时了三种常见的任务创建方式，并且创建任务是可以指定任务创建的选项，从而达到最优的创建方式。

在TaskCreationOptions中一共有7个枚举，枚举是可以使用|运算符组合定义的。其枚举如下表所示。

成员名称	说明
AttachedToParent	指定将任务附加到任务层次结构中的某个父级。 默认情况下，子任务（即由外部任务创建的内部任务）将独立于其父任务执行。 可以使用 TaskContinuationOptions.AttachedToParent 选项以便将父任务和子任务同步。请注意，如果使用 DenyChildAttach 选项配置父任务，则子任务中的 AttachedToParent 选项不起作用，并且子任务将作为分离的子任务执行。有关详细信息，请参阅附加和分离的子任务。
DenyChildAttach	指定任何尝试作为附加的子任务执行（即，使用 AttachedToParent 选项创建）的子任务都无法附加到父任务，会改成作为分离的子任务执行。 有关详细信息，请参阅附加和分离的子任务。
HideScheduler	防止环境计划程序被视为已创建任务的当前计划程序。 这意味着像 StartNew 或 ContinueWith 创建任务的执行操作将被视为 Default 当前计划程序。
LongRunning	指定任务将是长时间运行的、粗粒度的操作，涉及比细化的系统更少、更大的组件。 它会向 TaskScheduler 提示，过度订阅可能是合理的。 可以通过过度订阅创建比可用硬件线程数更多的线程。 它还将提示任务计划程序：该任务需要附加线程，以使任务不阻塞本地线程池队列中其他线程或工作项的向前推动。
None	指定应使用默认行为。
PreferFairness	提示 TaskScheduler 以一种尽可能公平的方式安排任务，这意味着较早安排的任务将更可能较早运行，而较晚安排运行的任务将更可能较晚运行。
RunContinuationsAsynchronously	强制异步执行添加到当前任务的延续任务。请注意，RunContinuationsAsynchronously 成员在以 .NET Framework 4.6 开头的 TaskCreationOptions 枚举中可用。

Copy

static void Main(string[] args) { // 使用构造方法创建任务 var t1 = new Task(() => TaskMethod("Task 1")); var t2 = new Task(() => TaskMethod("Task 2")); // 需要手动启动 t2.Start(); t1.Start(); // 使用Task.Run 方法启动任务 不需要手动启动 Task.Run(() => TaskMethod("Task 3")); // 使用 Task.Factory.StartNew方法 启动任务 实际上就是Task.Run Task.Factory.StartNew(() => TaskMethod("Task 4")); // 在StartNew的基础上 添加 TaskCreationOptions.LongRunning 告诉 Factory该任务需要长时间运行 // 那么它就会可能会创建一个 非线程池线程来执行任务 Task.Factory.StartNew(() => TaskMethod("Task 5"), TaskCreationOptions.LongRunning); ReadLine(); } static void TaskMethod(string name) { WriteLine($"任务 {name} 运行，线程 id {CurrentThread.ManagedThreadId}. 是否为线程池线程: {CurrentThread.IsThreadPoolThread}."); }

运行结果如下图所示。

1.3 使用任务执行基本的操作#

在本节中，使用任务执行基本的操作，并且获取任务执行完成后的结果值。本节内容比较简单，在此不做过多介绍。

演示代码如下，在主线程中要获取结果值，常用的方式就是访问task.Result属性，如果任务线程还没执行完毕，那么会阻塞主线程，直到线程执行完。如果任务线程执行完毕，那么将直接拿到运算的结果值。

在Task 3中，使用了task.Status来打印线程的状态，线程每个状态的具体含义，将在下一节中介绍。

Copy

static void Main(string[] args) { // 直接执行方法 作为参照 TaskMethod("主线程任务"); // 访问 Result属性 达到运行结果 Task<int> task = CreateTask("Task 1"); task.Start(); int result = task.Result; WriteLine($"运算结果: {result}"); // 使用当前线程，同步执行任务 task = CreateTask("Task 2"); task.RunSynchronously(); result = task.Result; WriteLine($"运算结果：{result}"); // 通过循环等待 获取运行结果 task = CreateTask("Task 3"); WriteLine(task.Status); task.Start(); while (!task.IsCompleted) { WriteLine(task.Status); Sleep(TimeSpan.FromSeconds(0.5)); } WriteLine(task.Status); result = task.Result; WriteLine($"运算结果：{result}"); Console.ReadLine(); } static Task<int> CreateTask(string name) { return new Task<int>(() => TaskMethod(name)); } static int TaskMethod(string name) { WriteLine($"{name} 运行在线程 {CurrentThread.ManagedThreadId}上. 是否为线程池线程 {CurrentThread.IsThreadPoolThread}"); Sleep(TimeSpan.FromSeconds(2)); return 42; }

运行结果如下，可见Task 1 和Task 2均是运行在主线程上，并非线程池线程。

1.4 组合任务#

在本节中，体现了任务其中一个强大的功能，那就是组合任务。通过组合任务可很好的描述任务与任务之间的异步、同步关系，大大降低了编程的难度。

组合任务主要是通过task.ContinueWith()、task.WhenAny()、task.WhenAll()等和task.GetAwaiter().OnCompleted()方法来实现。

在使用task.ContinueWith()方法时，需要注意它也可传递一系列的枚举选项TaskContinuationOptions，该枚举选项和TaskCreationOptions类似，其具体定义如下表所示。

成员名称	说明
AttachedToParent	如果延续为子任务，则指定将延续附加到任务层次结构中的父级。 只有当延续前面的任务也是子任务时，延续才可以是子任务。 默认情况下，子任务（即由外部任务创建的内部任务）将独立于其父任务执行。 可以使用 TaskContinuationOptions.AttachedToParent 选项以便将父任务和子任务同步。请注意，如果使用 DenyChildAttach 选项配置父任务，则子任务中的 AttachedToParent 选项不起作用，并且子任务将作为分离的子任务执行。有关更多信息，请参见Attached and Detached Child Tasks。
DenyChildAttach	指定任何使用 TaskCreationOptions.AttachedToParent 选项创建，并尝试作为附加的子任务执行的子任务（即，由此延续创建的任何嵌套内部任务）都无法附加到父任务，会改成作为分离的子任务执行。 有关详细信息，请参阅附加和分离的子任务。
ExecuteSynchronously	指定应同步执行延续任务。 指定此选项后，延续任务在导致前面的任务转换为其最终状态的相同线程上运行。如果在创建延续任务时已经完成前面的任务，则延续任务将在创建此延续任务的线程上运行。 如果前面任务的 CancellationTokenSource 已在一个 finally（在 Visual Basic 中为 Finally）块中释放，则使用此选项的延续任务将在该 finally 块中运行。 只应同步执行运行时间非常短的延续任务。由于任务以同步方式执行，因此无需调用诸如 Task.Wait 的方法来确保调用线程等待任务完成。
HideScheduler	指定由延续通过调用方法（如 Task.Run 或 Task.ContinueWith）创建的任务将默认计划程序 (TaskScheduler.Default) 视为当前的计划程序，而不是正在运行该延续的计划程序。
LazyCancellation	在延续取消的情况下，防止延续的完成直到完成先前的任务。
LongRunning	指定延续将是长期运行的、粗粒度的操作。 它会向 TaskScheduler 提示，过度订阅可能是合理的。
None	如果未指定延续选项，应在执行延续任务时使用指定的默认行为。 延续任务在前面的任务完成后以异步方式运行，与前面任务最终的 Task.Status 属性值无关。 如果延续为子任务，则会将其创建为分离的嵌套任务。
NotOnCanceled	指定不应在延续任务前面的任务已取消的情况下安排延续任务。 如果前面任务完成的 Task.Status 属性是 TaskStatus.Canceled，则前面的任务会取消。 此选项对多任务延续无效。
NotOnFaulted	指定不应在延续任务前面的任务引发了未处理异常的情况下安排延续任务。 如果前面任务完成的 Task.Status 属性是 TaskStatus.Faulted，则前面的任务会引发未处理的异常。 此选项对多任务延续无效。
NotOnRanToCompletion	指定不应在延续任务前面的任务已完成运行的情况下安排延续任务。 如果前面任务完成的 Task.Status 属性是 TaskStatus.RanToCompletion，则前面的任务会运行直至完成。 此选项对多任务延续无效。
OnlyOnCanceled	指定只应在延续前面的任务已取消的情况下安排延续任务。 如果前面任务完成的 Task.Status 属性是 TaskStatus.Canceled，则前面的任务会取消。 此选项对多任务延续无效。
OnlyOnFaulted	指定只有在延续任务前面的任务引发了未处理异常的情况下才应安排延续任务。 如果前面任务完成的 Task.Status 属性是 TaskStatus.Faulted，则前面的任务会引发未处理的异常。OnlyOnFaulted 选项可保证前面任务中的 Task.Exception 属性不是 null。 你可以使用该属性来捕获异常，并确定导致任务出错的异常。 如果你不访问 Exception 属性，则不会处理异常。 此外，如果尝试访问已取消或出错的任务的 Result 属性，则会引发一个新异常。此选项对多任务延续无效。
OnlyOnRanToCompletion	指定只应在延续任务前面的任务已完成运行的情况下才安排延续任务。 如果前面任务完成的 Task.Status 属性是 TaskStatus.RanToCompletion，则前面的任务会运行直至完成。 此选项对多任务延续无效。
PreferFairness	提示 TaskScheduler 按任务计划的顺序安排任务，因此较早安排的任务将更可能较早运行，而较晚安排运行的任务将更可能较晚运行。
RunContinuationsAsynchronously	指定应异步运行延续任务。 此选项优先于 TaskContinuationOptions.ExecuteSynchronously。

演示代码如下所示，使用ContinueWith()和OnCompleted()方法组合了任务来运行，搭配不同的TaskCreationOptions和TaskContinuationOptions来实现不同的效果。

Copy

static void Main(string[] args) { WriteLine($"主线程 线程 Id {CurrentThread.ManagedThreadId}"); // 创建两个任务 var firstTask = new Task<int>(() => TaskMethod("Frist Task",3)); var secondTask = new Task<int>(()=> TaskMethod("Second Task",2)); // 在默认的情况下 ContiueWith会在前面任务运行后再运行 firstTask.ContinueWith(t => WriteLine($"第一次运行答案是 {t.Result}. 线程Id {CurrentThread.ManagedThreadId}. 是否为线程池线程: {CurrentThread.IsThreadPoolThread}")); // 启动任务 firstTask.Start(); secondTask.Start(); Sleep(TimeSpan.FromSeconds(4)); // 这里会紧接着 Second Task运行后运行， 但是由于添加了 OnlyOnRanToCompletion 和 ExecuteSynchronously 所以会由运行SecondTask的线程来 运行这个任务 Task continuation = secondTask.ContinueWith(t => WriteLine($"第二次运行的答案是 {t.Result}. 线程Id {CurrentThread.ManagedThreadId}. 是否为线程池线程：{CurrentThread.IsThreadPoolThread}"),TaskContinuationOptions.OnlyOnRanToCompletion | TaskContinuationOptions.ExecuteSynchronously); // OnCompleted 是一个事件 当contiuation运行完成后 执行OnCompleted Action事件 continuation.GetAwaiter().OnCompleted(() => WriteLine($"后继任务完成. 线程Id {CurrentThread.ManagedThreadId}. 是否为线程池线程 {CurrentThread.IsThreadPoolThread}")); Sleep(TimeSpan.FromSeconds(2)); WriteLine(); firstTask = new Task<int>(() => { // 使用了TaskCreationOptions.AttachedToParent 将这个Task和父Task关联， 当这个Task没有结束时 父Task 状态为 WaitingForChildrenToComplete var innerTask = Task.Factory.StartNew(() => TaskMethod("Second Task",5), TaskCreationOptions.AttachedToParent); innerTask.ContinueWith(t => TaskMethod("Thrid Task", 2), TaskContinuationOptions.AttachedToParent); return TaskMethod("First Task",2); }); firstTask.Start(); // 检查firstTask线程状态 根据上面的分析 首先是 Running -> WatingForChildrenToComplete -> RanToCompletion while (! firstTask.IsCompleted) { WriteLine(firstTask.Status); Sleep(TimeSpan.FromSeconds(0.5)); } WriteLine(firstTask.Status); Console.ReadLine(); } static int TaskMethod(string name, int seconds) { WriteLine($"任务 {name} 正在运行,线程池线程 Id {CurrentThread.ManagedThreadId},是否为线程池线程: {CurrentThread.IsThreadPoolThread}"); Sleep(TimeSpan.FromSeconds(seconds)); return 42 * seconds; }

运行结果如下图所示，与预期结果一致。其中使用了task.Status来打印任务运行的状态，对于task.Status的状态具体含义如下表所示。

成员名称	说明
Canceled	该任务已通过对其自身的 CancellationToken 引发 OperationCanceledException 对取消进行了确认，此时该标记处于已发送信号状态；或者在该任务开始执行之前，已向该任务的 CancellationToken 发出了信号。 有关详细信息，请参阅任务取消。
Created	该任务已初始化，但尚未被计划。
Faulted	由于未处理异常的原因而完成的任务。
RanToCompletion	已成功完成执行的任务。
Running	该任务正在运行，但尚未完成。
WaitingForActivation	该任务正在等待 .NET Framework 基础结构在内部将其激活并进行计划。
WaitingForChildrenToComplete	该任务已完成执行，正在隐式等待附加的子任务完成。
WaitingToRun	该任务已被计划执行，但尚未开始执行。

1.5 将APM模式转换为任务#

在前面的章节中，介绍了基于IAsyncResult接口实现了BeginXXXX/EndXXXX方法的就叫APM模式。APM模式非常古老，那么如何将它转换为TAP模式呢？对于常见的几种APM模式异步任务，我们一般选择使用Task.Factory.FromAsync()方法来实现将APM模式转换为TAP模式。

演示代码如下所示，比较简单不作过多介绍。

Copy

static void Main(string[] args) { int threadId; AsynchronousTask d = Test; IncompatibleAsychronousTask e = Test; // 使用 Task.Factory.FromAsync方法 转换为Task WriteLine("Option 1"); Task<string> task = Task<string>.Factory.FromAsync(d.BeginInvoke("异步任务线程", CallBack, "委托异步调用"), d.EndInvoke); task.ContinueWith(t => WriteLine($"回调函数执行完毕，现在运行续接函数！结果：{t.Result}")); while (!task.IsCompleted) { WriteLine(task.Status); Sleep(TimeSpan.FromSeconds(0.5)); } WriteLine(task.Status); Sleep(TimeSpan.FromSeconds(1)); WriteLine("----------------------------------------------"); WriteLine(); // 使用 Task.Factory.FromAsync重载方法 转换为Task WriteLine("Option 2"); task = Task<string>.Factory.FromAsync(d.BeginInvoke,d.EndInvoke,"异步任务线程","委托异步调用"); task.ContinueWith(t => WriteLine($"任务完成，现在运行续接函数！结果：{t.Result}")); while (!task.IsCompleted) { WriteLine(task.Status); Sleep(TimeSpan.FromSeconds(0.5)); } WriteLine(task.Status); Sleep(TimeSpan.FromSeconds(1)); WriteLine("----------------------------------------------"); WriteLine(); // 同样可以使用 FromAsync方法 将 BeginInvoke 转换为 IAsyncResult 最后转换为 Task WriteLine("Option 3"); IAsyncResult ar = e.BeginInvoke(out threadId, CallBack, "委托异步调用"); task = Task<string>.Factory.FromAsync(ar, _ => e.EndInvoke(out threadId, ar)); task.ContinueWith(t => WriteLine($"任务完成，现在运行续接函数！结果：{t.Result}，线程Id {threadId}")); while (!task.IsCompleted) { WriteLine(task.Status); Sleep(TimeSpan.FromSeconds(0.5)); } WriteLine(task.Status); ReadLine(); } delegate string AsynchronousTask(string threadName); delegate string IncompatibleAsychronousTask(out int threadId); static void CallBack(IAsyncResult ar) { WriteLine("开始运行回调函数..."); WriteLine($"传递给回调函数的状态{ar.AsyncState}"); WriteLine($"是否为线程池线程：{CurrentThread.IsThreadPoolThread}"); WriteLine($"线程池工作线程Id：{CurrentThread.ManagedThreadId}"); } static string Test(string threadName) { WriteLine("开始运行..."); WriteLine($"是否为线程池线程：{CurrentThread.IsThreadPoolThread}"); Sleep(TimeSpan.FromSeconds(2)); CurrentThread.Name = threadName; return $"线程名：{CurrentThread.Name}"; } static string Test(out int threadId) { WriteLine("开始运行..."); WriteLine($"是否为线程池线程：{CurrentThread.IsThreadPoolThread}"); Sleep(TimeSpan.FromSeconds(2)); threadId = CurrentThread.ManagedThreadId; return $"线程池线程工作Id是：{threadId}"; }

运行结果如下图所示。

1.6 将EAP模式转换为任务#

在上几章中有提到，通过BackgroundWorker类通过事件的方式实现的异步，我们叫它EAP模式。那么如何将EAP模式转换为任务呢？很简单，我们只需要通过TaskCompletionSource类，即可将EAP模式转换为任务。

演示代码如下所示。

Copy

static void Main(string[] args) { var tcs = new TaskCompletionSource<int>(); var worker = new BackgroundWorker(); worker.DoWork += (sender, eventArgs) => { eventArgs.Result = TaskMethod("后台工作", 5); }; // 通过此方法 将EAP模式转换为 任务 worker.RunWorkerCompleted += (sender, eventArgs) => { if (eventArgs.Error != null) { tcs.SetException(eventArgs.Error); } else if (eventArgs.Cancelled) { tcs.SetCanceled(); } else { tcs.SetResult((int)eventArgs.Result); } }; worker.RunWorkerAsync(); // 调用结果 int result = tcs.Task.Result; WriteLine($"结果是：{result}"); ReadLine(); } static int TaskMethod(string name, int seconds) { WriteLine($"任务{name}运行在线程{CurrentThread.ManagedThreadId}上. 是否为线程池线程{CurrentThread.IsThreadPoolThread}"); Sleep(TimeSpan.FromSeconds(seconds)); return 42 * seconds; }

运行结果如下图所示。

1.7 实现取消选项#

在TAP模式中，实现取消选项和之前的异步模式一样，都是使用CancellationToken来实现，但是不同的是Task构造函数允许传入一个CancellationToken，从而在任务实际启动之前取消它。

演示代码如下所示。

Copy

static void Main(string[] args) { var cts = new CancellationTokenSource(); // new Task时 可以传入一个 CancellationToken对象 可以在线程创建时 变取消任务 var longTask = new Task<int>(() => TaskMethod("Task 1", 10, cts.Token), cts.Token); WriteLine(longTask.Status); cts.Cancel(); WriteLine(longTask.Status); WriteLine("第一个任务在运行前被取消."); // 同样的 可以通过CancellationToken对象 取消正在运行的任务 cts = new CancellationTokenSource(); longTask = new Task<int>(() => TaskMethod("Task 2", 10, cts.Token), cts.Token); longTask.Start(); for (int i = 0; i < 5; i++) { Sleep(TimeSpan.FromSeconds(0.5)); WriteLine(longTask.Status); } cts.Cancel(); for (int i = 0; i < 5; i++) { Sleep(TimeSpan.FromSeconds(0.5)); WriteLine(longTask.Status); } WriteLine($"这个任务已完成，结果为{longTask.Result}"); ReadLine(); } static int TaskMethod(string name, int seconds, CancellationToken token) { WriteLine($"任务运行在{CurrentThread.ManagedThreadId}上. 是否为线程池线程：{CurrentThread.IsThreadPoolThread}"); for (int i = 0; i < seconds; i++) { Sleep(TimeSpan.FromSeconds(1)); if (token.IsCancellationRequested) { return -1; } } return 42 * seconds; }

运行结果如下图所示，这里需要注意的是，如果是在任务执行之前取消了任务，那么它的最终状态是Canceled。如果是在执行过程中取消任务，那么它的状态是RanCompletion。

1.8 处理任务中的异常#

在任务中，处理异常和其它异步方式处理异常类似，如果能在所发生异常的线程中处理，那么不要在其它地方处理。但是对于一些不可预料的异常，那么可以通过几种方式来处理。

可以通过访问task.Result属性来处理异常，因为访问这个属性的Get方法会使当前线程等待直到该任务完成，并将异常传播给当前线程，这样就可以通过try catch语句块来捕获异常。另外使用task.GetAwaiter().GetResult()方法和第使用task.Result类似，同样可以捕获异常。如果是要捕获多个任务中的异常错误，那么可以通过ContinueWith()方法来处理。

具体如何实现，演示代码如下所示。

Copy

static void Main(string[] args) { Task<int> task; // 在主线程中调用 task.Result task中的异常信息会直接抛出到 主线程中 try { task = Task.Run(() => TaskMethod("Task 1", 2)); int result = task.Result; WriteLine($"结果为: {result}"); } catch (Exception ex) { WriteLine($"异常被捕捉：{ex.Message}"); } WriteLine("------------------------------------------------"); WriteLine(); // 同上 只是访问Result的方式不同 try { task = Task.Run(() => TaskMethod("Task 2", 2)); int result = task.GetAwaiter().GetResult(); WriteLine($"结果为：{result}"); } catch (Exception ex) { WriteLine($"异常被捕捉: {ex.Message}"); } WriteLine("----------------------------------------------"); WriteLine(); var t1 = new Task<int>(() => TaskMethod("Task 3", 3)); var t2 = new Task<int