关于 Task 简单梳理

〇、前言

Task 是微软在 .Net 4.0 时代推出来的，也是微软极力推荐的一种多线程的处理方式。

在 Task 之前有一个高效多线程操作累 ThreadPool，虽然线程池相对于 Thread，具有很多优势避免频繁创建和销毁线程等，但是线程池也有一些使用上的不便，比如不支持取消、完成、失败通知等，也不支持线程执行的先后顺序配置。

为了解决上述痛点，Task 诞生了。Task 就是站在巨人的肩膀上而生，它是基于 ThreadPool 封装。Task 的控制和扩展性很强，在线程的延续、阻塞、取消、超时等方面远胜于 ThreadPool。

本文将对 Task 进行一个详细的介绍。

一、任务如何创建和启动？

创建任务和执行任务是可以分离的，也可以同时进行。如下代码有四种开启任务的方式：

• 第一种：任务 t1 通过调用 Task 类构造函数进行实例化，但仅在任务 t2 启动后调用其 Start() 方法启动。【创建+未启动】
• 第二种：任务 t2 通过调用 TaskFactory.StartNew(Action<Object>, Object) 方法在单个方法调用中实例化和启动。【创建+启动】
• 第三种：任务 t3 通过调用 Run(Action) 方法在单个方法调用中实例化和启动。【创建+启动】
• 第四种：任务 t4 通过调用 RunSynchronously() 方法在主线程上同步执行。【创建+未启动】

static void Main(string[] args)
{
    // 用于异步调用的委托函数，接受类型为 Object 的参数
    Action<object> action = (object obj) =>
    {
        // Task.CurrentId ：任务 ID
        // Thread.CurrentThread.ManagedThreadId ：线程 ID
        Console.WriteLine($"Task={Task.CurrentId}, obj={obj}, Thread={Thread.CurrentThread.ManagedThreadId}");
        // throw new Exception();
    };

    // 【第一种】创建一个就绪，但【未启动】的任务，需要在后文通过 t1.Start() 启动
    Task t1 = new Task(action, "甲"); // alpha：初始

    // 【第二种】【创建并启动】一个任务
    Task t2 = Task.Factory.StartNew(action, "乙");
    // 占用主线程，等待任务 t2 完成
    t2.Wait();

    // 启动第一个任务 t1
    t1.Start();
    Console.WriteLine($"t1 已启动  (主线程 = {Thread.CurrentThread.ManagedThreadId})");
    // 通过 Wait() 占用主线程，等待 t1 执行完毕
    t1.Wait();

    // 【第三种】通过 Task.Run() 【创建并启动】一个任务
    string taskData = "丙";
    Task t3 = Task.Run(() =>
    {
        Console.WriteLine($"Task={Task.CurrentId}, obj={taskData}, Thread={Thread.CurrentThread.ManagedThreadId}");
    });
    // 通过 Wait() 占用主线程，等待 t3 执行完毕
    t3.Wait();

    // 【第四种】创建一个就绪，但【未启动】的任务 t4
    Task t4 = new Task(action, "丁");
    // Synchronously：同步的
    // 开启同步任务 t4，在主线程上运行
    t4.RunSynchronously();
    // t4 是以同步的方式运行的，此时的 Wait() 可以捕捉到异常
    t4.Wait();
    Console.ReadLine();
}

如下图输出结果，最先开启的 t2，由于是工厂中启动的，所以不占用主线程运行。Task.Run() 同样是非主线程运行，但它并未新开线程，而是直接用了 t2 执行的线程。

线程编号为 1 的是主线程，t1 是主线程最先创建的，所以直接由主线程运行。t4 是在同步执行的任务，因此也是主线程来执行。

二、等待一个或多个任务

用于等待任务的方法有很多个，如下：

下面几个示例来实操下。

2.1 Wait()

对于 Wait() 单线程等待，没啥好说的，看代码：

static void Main(string[] args)
{
    // 创建并执行一个任务执行匿名函数
    Task taskA = Task.Run(() => Thread.Sleep(2000));
    Console.WriteLine($"taskA Status: {taskA.Status}"); // taskA Status: WaitingToRun
    try
    {
        taskA.Wait(1000); // 主线程等待任务 1s 此时任务尚未完成
        Console.WriteLine($"taskA Status: {taskA.Status}"); // taskA Status: Running
        taskA.Wait(); // 线程等待任务 taskA 完成
        Console.WriteLine($"taskA Status: {taskA.Status}"); // taskA Status: RanToCompletion
    }
    catch (AggregateException)
    {
        Console.WriteLine("Exception in taskA.");
    }
}

2.2 Wait(Int32, CancellationToken)  支持手动取消

关于 Wait(Int32, CancellationToken) 任务可手动取消的重载。在任务完成之前，超时或调用了 Cancel() 方法，等待终止。

如下示例，一个线程一个任务，线程中将 CabcellationTokenSource 的实例 cts 取消掉，导致后续任务等待时调用 cts.Token 导致异常 OperationCanceledException 的发生。

static void Main(string[] args)
{
    CancellationTokenSource cts = new CancellationTokenSource();
    Thread thread = new Thread(CancelToken); // 新开一个线程执行方法：CancelToken()
    thread.Start(cts);

    Task t = Task.Run(() => // 新增一个任务执行匿名函数
    {
        Task.Delay(5000).Wait(); // 延迟等待 5s
        Console.WriteLine("Task ended delay...");
    });
    try
    {
        Console.WriteLine($"About to wait completion of task {t.Id}"); // 以上两个操作都有延迟，所以此处消息先打印
        // 等待任务 t 1.51s，保证线程已执行完成，就是保证 CancellationTokenSource 已执行过取消操作
        // 由于 cts 已经取消，因此次数就抛异常：OperationCanceledException
        bool result = t.Wait(1510, cts.Token); // 后边代码就不再执行，直接跳到 catch
        Console.WriteLine($"Wait completed normally: {result}");
        Console.WriteLine($"The task status:  {t.Status}");
    }
    catch (OperationCanceledException e)
    {
        Console.WriteLine($"{e.GetType().Name}: The wait has been canceled.");
        Console.WriteLine($"Task status：{t.Status}"); // 此时程序运行 1.5s 多，任务 t 还在等待，因此状态是 Running
        Thread.Sleep(4000); // 4s + 1.5s > 5s 此时任务 t 已经执行完成，状态为 RanToCompletion 
        Console.WriteLine("After sleeping, the task status:  {t.Status}");
        cts.Dispose();
    }
    Console.ReadLine();
}

private static void CancelToken(Object obj)
{
    Thread.Sleep(1500); // 延迟 1.5s
    Console.WriteLine($"Canceling the cancellation token from thread {Thread.CurrentThread.ManagedThreadId}...");
    CancellationTokenSource source = obj as CancellationTokenSource;
    if (source != null) 
        source.Cancel(); // 将 CancellationTokenSource 的实例执行取消
}

2.3 WaitAll()

 等待一组任务全部完成，无论是否抛异常。AggregateException 将会收集全部异常信息，可以通过遍历获取每一个异常详情。

如下代码，新建是个任务组成任务组 tasks，其中 2~5 线程手动抛异常，最后通过遍历 AggregateException aex 记录全部异常。

static void Main(string[] args)
{
    var tasks = new List<Task<int>>();
    // 创建一个委托，用于任务执行，并记录每个任务信息
    Func<object, int> action = (object obj) =>
    {
        int i = (int)obj;
        // 让每次的 TickCount 不同（系统开始运行的毫秒数）
        Thread.Sleep(i * 1000);
        if (2 <= i && i <= 5) // 从第 2 到 5 个任务都抛异常
        {
            throw new InvalidOperationException("SIMULATED EXCEPTION");
        }
        int tickCount = Environment.TickCount; // 获取系统开始运行的毫秒数
        Console.WriteLine($"Task={Task.CurrentId}, i={i}, TickCount={tickCount}, Thread={Thread.CurrentThread.ManagedThreadId}");
        return tickCount;
    };
    // 连续创建 10 个任务
    for (int i = 0; i < 10; i++)
    {
        int index = i;
        tasks.Add(Task<int>.Factory.StartNew(action, index)); // 后台线程
    }
    try
    {
        // WaitAll() 等待全部任务完成
        Task.WaitAll(tasks.ToArray());
        // 由于线程中手动抛出了异常，因此这个消息将无法打印在控制台
        Console.WriteLine("WaitAll() has not thrown exceptions. THIS WAS NOT EXPECTED.");
    }
    catch (AggregateException aex) // AggregateException 异常中包含 2~5 四个异常
    {
        Console.WriteLine("\nThe following exceptions have been thrown by WaitAll(): (THIS WAS EXPECTED)");
        Console.WriteLine($"\ne.InnerExceptions.Count：{aex.InnerExceptions.Count}");
        for (int j = 0; j < aex.InnerExceptions.Count; j++) // aex.InnerExceptions.Count == 4
        {
            Console.WriteLine("\n-------------------------------------------------\n{0}", aex.InnerExceptions[j].ToString());
        }
    }
    Console.ReadLine();
}

2.4 WaitAny()

 等待一组任务中的任一任务完成，然后返回第一个执行完成任务的序号，可通过tasks[index].Id取得任务 ID。

如下示例，每个任务都有延迟，当第一个任务完成时，遍历打印出其他全部任务的状态：

static void Main(string[] args)
{
    Task[] tasks = new Task[5];
    for (int ctr = 0; ctr <= 4; ctr++)
    {
        int factor = ctr; // 重新声明一个变量
        tasks[ctr] = Task.Run(() => Thread.Sleep(factor * 250 + 50));
    }
    int index = Task.WaitAny(tasks); // 等待任一任务结束
    Console.WriteLine($"任务 #{tasks[index].Id} 已完成。");
    Console.WriteLine("\n当前各个任务的状态：");
    foreach (var t in tasks)
        Console.WriteLine($"   Task {t.Id}: {t.Status}");
    Console.ReadLine();
}

参考：https://learn.microsoft.com/zh-cn/dotnet/api/system.threading.tasks.task.wait?view=net-7.0

三、延续任务 Task.ContinueWith()

3.1 一个简单的示例

如下代码，首先创建一个耗时的任务 task 并启动，此时也不影响主线程的运行。然后通过task.ContinueWith()在第一个任务执行完成后，执行其中的匿名函数。

static void Main(string[] args)
{
    // 创建一个任务
    Task<int> task = new Task<int>(() =>
    {
        int sum = 0;
        Console.WriteLine($"使用 Task 执行异步操作，当前线程 {Thread.CurrentThread.ManagedThreadId}");
        Thread.Sleep(2000);
        for (int i = 0; i < 100; i++)
        {
            sum += i;
        }
        return sum;
    });
    // 启动任务
    task.Start();
    // 主线程在此处可以执行其他处理
    Console.WriteLine($"1 主线程 {Thread.CurrentThread.ManagedThreadId}");
    Thread.Sleep(1000);

    //任务完成时执行处理。
    Task cwt = task.ContinueWith(t =>
    {
        Console.WriteLine($"任务完成后的执行结果：{t.Result} 当前线程 {Thread.CurrentThread.ManagedThreadId}");
    });
    task.Wait();
    cwt.Wait();
    Console.WriteLine($"2 主线程 {Thread.CurrentThread.ManagedThreadId}");
    Console.ReadLine();
}

详情可参考：https://learn.microsoft.com/zh-cn/dotnet/api/system.threading.tasks.task.continuewith?view=net-7.0

3.2 任务的并行与串行

ContinueWith、WaitAll 当这两者结合起来，我们就可以处理复杂一点的东西。比如，现在有 7 个任务，其中 t1 需要串行，t2-t3 可以并行，t4 需要串行，t5-t6 并行，t7 串行。逻辑如下图：

public static void Main(string[] args)
{
    ConcurrentStack<int> stack = new ConcurrentStack<int>(); // ConcurrentStack：线程安全的后进先出(LIFO：LastIn-FirstOut)集合
    ConcurrentBag<int> bag = new ConcurrentBag<int>(); // ConcurrentBag：线程安全的无序集合
    // t1先串行
    var t1 = Task.Factory.StartNew(() =>
    {
        stack.Push(1);
        stack.Push(2);
    });

    // t1.ContinueWith() t1 之后，t2、t3并行执行
    var t2 = t1.ContinueWith(t =>
    {
        int result;
        stack.TryPop(out result);
    });
    // t2,t3并行执行
    var t3 = t1.ContinueWith(t =>
    {
        int result;
        stack.TryPop(out result);
    });
    // 等待 t2、t3 执行完
    Task.WaitAll(t2, t3);

    //t4串行执行
    var t4 = Task.Factory.StartNew(() =>
    {
        stack.Push(1);
        stack.Push(2);
    });

    // t5、t6 并行执行
    var t5 = t4.ContinueWith(t =>
    {
        int result;
        stack.TryPop(out result);
    });
    // t5、t6 并行执行
    var t6 = t4.ContinueWith(t =>
    {
        int result;
        // 只弹出，不移除
        stack.TryPeek(out result);
    });
    // 临界区：等待 t5、t6 执行完
    Task.WaitAll(t5, t6);

    // t7 串行执行
    var t7 = Task.Factory.StartNew(() =>
    {
        Console.WriteLine($"当前集合元素个数：{stack.Count}"); // 当前集合元素个数：1
    });
    Console.ReadLine();
}

参考： https://www.cnblogs.com/huangxincheng/archive/2012/04/03/2430638.html

https://learn.microsoft.com/zh-cn/dotnet/api/system.threading.tasks.task?view=net-7.0

https://www.cnblogs.com/zhaoshujie/p/11082753.html

本栏目推荐文章
• [linux kernel] struct pid && the way the kernel used to cast uint32_t pid to struct task_struct
• celery中 @app.task与@shared.task 的区别
• 装载数据时报错：Failed to execute spark task, with exception 'org.apache.hadoop.hive.ql.metadata.HiveException(Failed to create spark client.)'
• C#中 同步方法调用异步方法不死锁的方法 async await task
• Spark任务存在大量Task失败记录分享
• 【WALT】top task 相关代码详解
• 【WALT】predict_and_update_buckets() 与 update_task_pred_demand() 代码详解
• swoole运行Master 进程、Reactor 线程、Worker 进程、Task 进程、Manager 进程的区别与联系
• 关于 Task 简单梳理（C#）【并发编程系列_4】
• C# Task详解

Tasktask1 task4 task2 task5 task3 multi-task task0 task6 create_task notes-task