C#多线程编程系列（五）- 使用任务并行库-白红宇

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

阅读量：791 次

发布时间：2023-01-23

本文共 6309 字，大约阅读时间需要 21 分钟。

.NET 框架 4.5 中任务并行库（TPL）学习笔记

1.1 简介

在之前的几章中，我们讨论了线程的使用和多线程开发中的相关内容。然而，使用多线程在项目中往往需要编写大量代码来保证正确性和性能。为了解决这些问题，.NET 框架 4.0 引入了任务并行库（TPL），该库在 4.5 版本中得到了改进。本文将基于最新版本的 TPL 库，详细介绍如何利用 TPL 来高效地处理多线程任务。

1.2 创建任务

任务是 TPL 中的核心概念，它是一个封装异步操作的对象。为了创建任务，可以使用 Task 类中的实例方法或静态方法。以下是几种常见的任务创建方式：

使用 `Task` 实例方法

var t1 = new Task(() => TaskMethod("Task 1"));
var t2 = new Task(() => TaskMethod("Task 2"));

使用 `Task.Run` 方法

Task.Run(() => TaskMethod("Task 3"));

使用 `Task.Factory.StartNew` 方法

Task.Factory.StartNew(() => TaskMethod("Task 4"));

使用 `TaskCreationOptions.LongRunning` 选项

Task.Factory.StartNew(() => TaskMethod("Task 5"), TaskCreationOptions.LongRunning);

任务创建选项

TaskCreationOptions 序列提供了多种配置选项，共计7种组合方式，具体说明如下：

1.3 使用任务执行基本操作

执行任务后，需要获取结果值。可以通过访问 task.Result 属性来获取结果，或在任务完成时通过 task.Wait() 和 task.Result 来获取结果。同时，任务有四种基本状态：Running、WaitingForActivation、WaitingForCompletion 和 WaitingForChildrenToComplete。

使用 `task.Result` 获取结果

task.Start();
int result = task.Result;

获取任务状态

WriteLine(task.Status); // 可能的状态包括 Running、WaitingForActivation、WaitingForCompletion等。

示例代码

static void Main(string[] args)
{
    var 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");
    while (!task.IsCompleted)
    {
        WriteLine(task.Status);
        Sleep(TimeSpan.FromSeconds(0.5));
    }
    WriteLine(task.Status);
    result = task.Result;
    WriteLine($"运算结果：{result}");
}

1.4 组合任务

任务组合是 TPL 的强大功能之一，用于定义任务之间的依赖关系或顺序执行。常用的方法包括 ContinueWith、WhenAny 和 WhenAll。

使用 `ContinueWith`

var firstTask = new Task(() => TaskMethod("First Task"));
var secondTask = new Task(() => TaskMethod("Second Task"));
firstTask.ContinueWith(t => WriteLine($"第一次运行答案是 {t.Result}."));

使用 `WhenAny` 和 `WhenAll`

Task.WhenAll(firstTask, secondTask).ContinueWith(t => WriteLine($"所有任务完成."));
Task.WhenAny(firstTask, secondTask).ContinueWith(t => WriteLine($"任意一个任务完成."));

1.5 将 APM 模式转换为 TAP 模式

APM（Asyncrhonous Programming Model）是一种基于 IAsyncResult 接口的异步模式。为了将其迁移到 TAP（Task Programming Model），可以使用 Task.Factory.FromAsync 方法。

示例代码

static void Main(string[] args)
{
    var d = Test("异步任务线程");
    var e = Test("IncompatibleAsychronousTask");
    
    WriteLine("Option 1: 使用 FromAsync 方法");
    var task = Task.Factory.FromAsync(d.BeginInvoke("异步任务线程", CallBack, "委托异步调用"), d.EndInvoke);
    task.ContinueWith(t => WriteLine($"回调函数执行完毕..."));
    
    WriteLine("Option 2: 使用从重载方法");
    task = Task.Factory.FromAsync(d.BeginInvoke, d.EndInvoke, "异步任务线程", "委托异步调用");
    task.ContinueWith(t => WriteLine($"任务完成..."));
    
    WriteLine("Option 3: 使用 FromAsync 方法处理 IAsyncResult");
    IAsyncResult ar = e.BeginInvoke(out int threadId, CallBack, "委托异步调用");
    var task = Task.Factory.FromAsync(ar, _ => e.EndInvoke(out threadId, ar));
    task.ContinueWith(t => WriteLine($"任务完成..."));
}

1.6 将 EAP 模式转换为 TAP 模式

EAP（Event-driven Asynchronous Pattern）通常使用 BackgroundWorker 类。为将其迁移到 TAP 模式，可以使用 TaskCompletionSource。

示例代码

static void Main(string[] args)
{
    var tcs = new TaskCompletionSource();
    var worker = new BackgroundWorker();
    worker.DoWork += (sender, eventArgs) => eventArgs.Result = TaskMethod("后台工作");
    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}");
}

1.7 实现取消选项

在 TAP 模式中，通过 CancellationToken 来实现取消功能。任务创建时可以传入 CancellationToken，在任务执行前如果取消，任务将以 Canceled 状态结束。

示例代码

static void Main(string[] args)
{
    var cts = new CancellationTokenSource();
    var longTask = new Task(() => TaskMethod("Task 1", 10), cts.Token);
    WriteLine(longTask.Status); // 运行中
    cts.Cancel();
    WriteLine(longTask.Status); // Canceled
}

1.8 处理任务中的异常

在任务中，异常可以通过 task.Result 应 eget 到，如果是在其他线程抛出的异常，则可以通过 ContinueWith 方法捕获并处理。

示例代码

static void Main(string[] args)
{
    try
    {
        var task = Task.Run(() => TaskMethod("Task 1", 2));
        int result = task.Result;
        WriteLine($"结果为: {result}");
    }
    catch (Exception ex)
    {
        WriteLine($"异常被捕捉: {ex.Message}");
    }
    
    var t1 = new Task(() => TaskMethod("Task 3", 3));
    var t2 = new Task(() => TaskMethod("Task 4", 4));
    var complexTask = Task.WhenAll(t1, t2);
    complexTask.ContinueWith(t => 
    {
        if (t.Exception != null)
        {
            WriteLine($"异常被捕捉: {t.Exception.Message}");
            foreach (Exception ex in t.Exception.InnerExceptions)
            {
                WriteLine($"-------------------------- {ex.Message}");
            }
        }
    }, TaskContinuationOptions.OnlyOnFaulted);
}

1.9 并行运行任务

TAP 提供了多种方法来并行运行任务。Task.WhenAll 用于等待所有任务完成，而 Task.WhenAny 则只需一个任务完成即可继续。

使用 `WhenAll`

var firstTask = new Task(() => TaskMethod("Task 1", 3));
var secondTask = new Task(() => TaskMethod("Task 2", 2));
var whenAllTask = Task.WhenAll(firstTask, secondTask);
whenAllTask.ContinueWith(t => WriteLine($"所有任务完成."));

使用 `WhenAny`

var tasks = new List
   
    ();
for (int i = 0; i < 4; i++)
{
    var task = new Task(() => TaskMethod($"Task {i + 1}", i + 1));
    tasks.Add(task);
}
var anyTask = Task.WhenAny(tasks);
while (tasks.Count > 0)
{
    var completedTask = anyTask.Result;
    tasks.Remove(completedTask);
    WriteLine($"一个任务已经完成...");
}

1.10 使用 TaskScheduler 配置任务执行

TaskScheduler 负责任务的调度。默认使用线程池任务调度器，但在 UI 组件中应使用同步上下文任务调度器，以避免跨线程更新 UI。

示例代码

static void Main(string[] args)
{
    MainWindow.CreateTextBlock("Sync", async () =>
    {
        await TaskMethod("任务同步执行");
    });
    
    MainWindow.CreateTextBlock("Async", async () =>
    {
        await TaskMethod("任务异步执行");
    });
    
    await TaskMethod("同步执行将导致 UI 线程阻塞");
}