C#多线程——任务并行库TPL

作者：contact97 时间：2024-03-07 阅读数：207 人阅读

1 什么是TPL

$\textcolor{red}{TPL}$ （Task Parallel Library）任务并行库，是从.NetFramwork4.0后引入的基于异步操作的一组API。TPL的底层是基于多线程实现的，但是它相较于直接使用多线程，更为简单，它向程序员隐藏了与线程池交互的底层代码。在.NetFramwork4.0后，微软更推荐程序员使用TPL去编写多线程代码或者并行代码。
TPL的核心是任务，一个任务代表了一个异步操作，该操作可以使用或不适用独立的线程运行。
一个任务可以和其他的任务组合起来，比如同时启动多个任务，等待所有的任务完成；对之前所有任务的结果进行计算，TPL的优势在于具有组合任务API，而不用单独书写线程同步的代码（关注于锁、线程间的信号）。同样在多线程中关于多线程中异常的传播与处理是极为复杂的，而在TPL中，可以通过 $\textcolor{red}{AggregateException}$ ，捕获底层任务的所有异常，并允许单独处理这些异常。

2 创建与启动任务

创建、启动任务使用到的关键类如下：类定义在 $\textcolor{red}{System.Threading.Tasks}$ 命名空间中
在这里插入图片描述

public static void Main(string[] args)
{
void TaskMethod(string name) {
Console.WriteLine("Task{0} is runing on ThreadId {1} Is ThreadPool Thread {2}",
name,
Thread.CurrentThread.ManagedThreadId,
Thread.CurrentThread.IsThreadPoolThread);
}

int TaskMethod_1(string name) {

Console.WriteLine("Task{0} is runing on ThreadId {1} Is ThreadPool Thread {2}",
name,
Thread.CurrentThread.ManagedThreadId,
Thread.CurrentThread.IsThreadPoolThread);
return 0;
}

#region 1 创建任务
TaskMethod("Main");
//public Task(Action action) 传入一个无返回的委托函数
Task t1 = new Task(() => TaskMethod("t1"));
Task t2 = new Task(() => TaskMethod("t2"));
t1.Start();//显示创建Task需要运行start方法才能运行任务中的方法
t2.RunSynchronously(); //RunSynchronously()在当前线程上运行任务方法
Task.Run(() => TaskMethod("t3"));
Task.Factory.StartNew(() => TaskMethod("t4"));

//Task<T> 管理有返回值的工作单元 public Task(Func<TResult> function)
//通过传入有返回的委托函数来创建任务对象
Task<int> t5 = new Task<int>(() => TaskMethod_1("t5"));
t5.Start();
Console.WriteLine(t5.Result); //获得任务结果

Console.ReadKey();
#endregion

}

在这里插入图片描述

3 等待任务

有以下方式可以等待任务完成：

调用Wait方法（可选择指定超时时间）
访问Result属性（当使用Task时）
Task.WaitAll（等待所有指定任务完成）
Task.WaitAny（等待任意一个任务完成）。

#region 2 等待任务

Console.OutputEncoding = Encoding.Unicode;
Console.WriteLine("起始执行时间为{0}", DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss"));
List<Task> list = new List<Task>();

for (int i = 1; i <= 5; i++)
{
int tempI = i; //需要使用局部变量，因为for循环在数据量很小的时候，for循环结束时 task启动了，但是可能还未执行。由于共享变量i，所有在真正执行Task时，线程名称将一样 i=11
Task t = new Task(()=>TaskMethod_2("-"+ tempI, tempI));
list.Add(t);
t.Start();//因为多线程的启动并不意味着立马进行，需要等待操作系统的调度。
}

Task.WaitAll(list.ToArray());//等待所有的线程完成
Task.WaitAny(list.ToArray());//等待任意一个线程完成后执行，相当于在一个ManualResetEventSlim上等待，

Console.WriteLine("等待所有任务线程执行完成,结束执行时间为{0}", DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss"));

Task<int> t_1 = new Task<int>(() => TaskMethod_2("t_1", 10));
t_1.Start();
t_1.Wait();//主线程等待t_1线程完成任务方法。 Wait(TimeSpan timeout) 也可以等待具体的时间
Console.WriteLine(t_1.Result);

Console.WriteLine("结束执行时间为{0}", DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss"));
Console.ReadKey();
#endregion

TaskMethod_2 方法如示例一代码块。方法运行后如下

在这里插入图片描述

4 任务中的异常处理

#region 3 任务的异常处理
Task t1 = Task.Factory.StartNew(() => {
throw new Exception("Task Failed ! ");
});
try
{
t1.Wait();//当你等待一个任务结束时（通过调用Wait方法或访问其Result属性），所有未处理的异常都会用一个AggregateException对象封装，方便重新抛给调用方。
}
catch (AggregateException aex)
{
Console.WriteLine(aex.InnerException.Message); // Task Failed !
}

//定义一个父子任务，在父任务与子任务中分别抛出异常
TaskCreationOptions atp = TaskCreationOptions.AttachedToParent;
var parent = Task.Factory.StartNew(() =>
{
Console.WriteLine("I am Parent");

Task.Factory.StartNew(() => // 子
{
Console.WriteLine("I am Child");
throw new Exception("Child Exception");
}, atp);

throw new Exception("Parent Exception");

});

try {
parent.Wait();
}
catch (AggregateException aex)
{
Console.WriteLine(aex.InnerExceptions.Count); //2 捕获了父异常和子异常
//这是对于有父子关系的任务，在父任务上等待也会隐式的等待子任务，所有子任务的
//异常也会传递出来。
}

Console.ReadKey();
#endregion

main 方法块如上，运行结果如下图所示
在这里插入图片描述

当某个任务抛出一个或多个异常时，异常包装在 $\textcolor{blue}{AggregateException }$ 异常中。该异常会传播回与任务联接的线程。通常，该线程是 $\textcolor{red}{等待任务完成}$ 的线程
（1）Wait方法
（2）WaitAny方法
（3）WaitAll方法
或访问 Result 属性的线程。AggregateException 通常包含关联任务线程中的所有异常，我们可以在外部通过try catch的方式去处理它，但是这并不意味着并不需要单独处理任务线程的异常，否则可能会因为无解的异常导致程式的中断。

5 取消任务

任务的取消需要用到 $\textcolor{red}{CancellationTokenSource}$ 和 $\textcolor{red}{CancellationToken}$ 类。在取消任务的过程中需要了解一下几点：

可以在构造中传入CancellationToken来构建Task任务，并且CancellationToken可以绑定到多个任务上
Task的创建和执行都是独立的，如果在任务执行前取消了任务，那么任务代码将不会执行。如果尝试调用start方法，将会抛出异常InvalidOperationException
任务执行后去执行CancellationTokenSource.Cancel方法，任务不会被取消
需要在任务代码中 $\textcolor{blue}{显示定义任务中断的逻辑}$

Main 方法代码块如下

#region 5 取消任务

Console.OutputEncoding = System.Text.Encoding.UTF8;
void RunTask(string name) {
for (int i = 0; i < 10; i++) {
Console.WriteLine(name+"运行开始:" + i);
Thread.Sleep(1000);
}
}

void RunTaskWithCancellationToken(string name,CancellationToken token) {
for (int i = 0; i < 10; i++)
{
Console.WriteLine(name + "运行开始:" + i);
Thread.Sleep(1000);
token.ThrowIfCancellationRequested(); //在运行的Task任务中，显示定义任务取消的逻辑
}
}

CancellationTokenSource source = new CancellationTokenSource();
CancellationToken cancellation = source.Token;

//创建一个和CancellationToken关联的任务类
//t1任务不支持显示取消任务
Task t1 = Task.Factory.StartNew(() => RunTask("t1"), cancellation);
//t2任务支持显示取消任务，在任务执行的逻辑中加了ThrowIfCancellationRequested，标识希望在任务中断是抛出异常
Task t2 = Task.Factory.StartNew(() => RunTaskWithCancellationToken("t2",cancellation), cancellation);
//t3任务不支持显示取消任务，且需要手动start去运行任务代码。
Task t3 = new Task(()=>RunTask("t1"),cancellation);

Thread.Sleep(TimeSpan.FromSeconds(3));

source.Cancel();

try
{
//在一个任务调度前，取消任务，那么将会抛出System.InvalidOperationException 标识在已经完成的工作上呼叫start动作。
t3.Start();
}
catch (Exception e)
{

Console.WriteLine("异常类型{0} ,异常消息{1}",e.GetType().Name,e.Message);
}

Console.WriteLine("t1 IsCanceled {0} ,t1 IsCompleted{1}, t1 IsFaulted{2}, t1 status{3}", t1.IsCanceled, t1.IsCompleted, t1.IsFaulted, t1.Status.ToString());
Console.WriteLine("t2 IsCanceled {0} ,t2 IsCompleted{1}, t2 IsFaulted{2}, t2 status{3}", t2.IsCanceled, t2.IsCompleted, t2.IsFaulted, t2.Status.ToString());
Console.WriteLine("t3 IsCanceled {0} ,t3 IsCompleted{1}, t3 IsFaulted{2}, t3 status{3}", t3.IsCanceled, t3.IsCompleted, t3.IsFaulted, t3.Status.ToString());
Console.ReadKey();
#endregion