《CLR via C#》之线程处理——线程池与任务

Posted 2020-08-04 乾学斋

《CLR via C#》之线程处理——线程池与任务

《CLR via C#》之线程处理——线程池与任务
线程池
线程池用法
任务
等待任务完成并获取结果(Wait方法和Result属性)
取消任务（token的ThrowIfCancellationRequested方法）
任务完成自动启动新任务（ContinueWith方法）
任务启动子任务
任务工厂（TaskFactory和TaskFactory）

线程池

每一个CLR都有一个线程池——由这个CLR控制的所有AppDomain共享。如果一个进程加载了多个CLR，那么每个CLR都有它自己的线程池。
CLR初始化化时，线程池是空的。在内部线程池维护了一个操作请求队列。应用请求异步操作时，就将一个记录项添加到队列中。线程池从队列中提取这个记录项，将这个记录项Dispatch给一个线程池线程。如果线程池没有线程，就创建一个新线程。当线程执行完后，也不会被销毁。
如果线程池闲着没事干一段时间之后，线程会终止自己以释放资源。

线程池用法

通常，调用ThreadPool类的静态方法：

static Boolean QueueUserWorkItem(WaitCallback callBack)；//callBack： 回调委托
static Boolean QueueUserWorkItem(WaitCallback callBa ck, Object state);//state：回调参数

任务

使用ThreadPool.QueueUserWorkItem方法有两个限制：

1. 没有内建机制通知操作何时完成；
2. 没有机制在操作完成时获得返回值。

因此，推出了Task来完成相同的事

ThreadPool.QueueUserWorkItem(ComputeBoundOp, 5);//调用QueueUserWorkItem
new Task(ComputeBoundOp, 5).Start();            //用Task做相同的事
Task.Run(() => ComputeBoundOp(5));              //等价写法

在构造Task时，可以传递TaskCreationOptions标识来控制Task的执行方式。

[Flags, Serializable]
public enum TaskCreationOptions
{
    None              = 0x0000, //默认
    // 提议TaskScheduler，希望尽快执行
    PreferFairness    = 0x0001,
    // 提议TaskScheduler，希望尽可能创建线程池线程
    LongRunning       = 0x0002,
    // 总被采纳：将Task和它的父Task关联
    AttachedToParent  = 0x0004,
    // 总被采纳：如果一个Task试图和父Task连接，它就是个普通Task，而不是子Task
    DenyChidAttach    = 0x0008,
    // 总被采纳：强迫子Task使用默认调度器而不是父Task的调度器
    HideScheduler     = 0x0010
}

等待任务完成并获取结果(Wait方法和Result属性)

    Task<int> t = new Task<int>(n => Sum((int)n), 100000000, TaskCreationOptions.LongRunning);
    t.Start();
    // 可选择显式等待任务完成
    t.Wait();
    // 可获得结果（Result属性内部会调用Wait）
    Console.WriteLine("The Sum is " + t.Result);

注意：Result属性内部会调用Wait方法。

---

取消任务（token的ThrowIfCancellationRequested方法）

Task和ThreadPool取消任务的区别是，前者在操作中调用用ThrowIfCancellationRequested抛出异常，后者检查IsCancellationRequested属性。原因是Task可以返回值，需要将已完成的任务和出错的任务区分开。

    internal static class TaskDemo
    {
        private static Int32 Sum(CancellationToken token, int n)
        {
            int sum = 0;
            for (; n > 0; n--)
            {
                // 如果token已取消，调用该方法会抛出一个OperationCanceledException
                token.ThrowIfCancellationRequested();
                checked
                {
                    sum += 10;
                }
            }
            return sum;
        }
        public static void Go()
        {
            CancellationTokenSource cts = new CancellationTokenSource();
            // 创建Task，并立即Start
            Task<int> t = Task.Run(() => Sum(cts.Token, 100000000), cts.Token);
            // 在之后的某个时间，取消task
            cts.Cancel();
            try
            {
                //如果任务已经取消，Result会抛出AggregateException
                Console.WriteLine("The sum is :" + t.Result);
            }
            catch(AggregateException x)
            {
                // 将任何OperationCancelException都视为已处理。
                // 其它任何异常都造成抛出一个新的AggregateException，其中只包含未处理的异常。
                x.Handle(e => e is OperationCanceledException);
                Console.WriteLine("Sum was cancelled");
            }
        }
    }

注意：如果试图取消一个未Start的任务，会抛出InvalidOperationException。

---

任务完成自动启动新任务（ContinueWith方法）

public static void TaskContinueWith()
{
    Task<int> t = Task.Run(() => Sum(10000));
    t.ContinueWith( task => Console.WriteLine("The sum is: " + task.Result),
        TaskContinuationOptions.OnlyOnRanToCompletion);
    t.ContinueWith(task => Console.WriteLine("Sum threw: " + task.Exception.InnerException),
        TaskContinuationOptions.OnlyOnFaulted);
    //ContinueWith返回Task，但一般都不需要保存它
    Task cwt = t.ContinueWith(task => Console.WriteLine("Sum was cancelled"),
        TaskContinuationOptions.OnlyOnCanceled);
}

TaskContinuationOptions枚举类型的前六个标识与TaskCreationOptions一致：

[Flags, Serializable]
public enum TaskContinuationOptions
{
    None                  = 0x0000,// 默认
    // 提议TaskScheduler，希望任务尽快执行
    PreferFairness        = 0x0001,
    // 提议TaskScheduler：应尽可能创建线程池线程
    LongRunning           = 0x0002,
    // 总被采纳：将一个Task与它的父Task关联
    AttachedToParent      = 0x0004,
    // 任务试图和这个父任务连接将抛出一个InvalidOperationException
    DenyChildAttach       = 0x0008,
    // 强迫子任务使用默认调度器而不是父任务的调度器
    HideScheduler         = 0x0010,
    // 除非前置任务完成，否则禁止延续任务完成（取消）
    LazyCancellation      = 0x0020,
    // 这个标识指出你希望由执行第一个任务的线程执行ContinueWith任务。
    // 第一个任务完成后，调用ContinueWith的线程接着执行ContinueWith任务。
    ExecuteSynchronously  = 0x80000,
    // 这些标识指出在什么情况下运行ContinueWith任务
    NotOnRanToCompletion  = 0x10000,
    NotOnFaulted          = 0x20000,
    NotOnCanceled         = 0x40000,
    // 这些标识是以上三个标识的便利组合
    OnlyOnCanceled        = NotOnRanToCompletion | NotOnFaulted,
    OnlyOnFaulted         = NotOnRanToCompletion | NotOnCanceled,
    OnlyOnRanToCompletion = NotOnFaulted | NotOnCanceled,
}

注意：ExecuteSynchronously：同步执行，两个任务在同一个线程一前一后地执行，被称为同步执行。

---

任务启动子任务

public static void TaskParentDemo()
{
    Task<Int32[]> parent = new Task<Int32[]>(() =>
    {
        var results = new Int32[3];    //创建一个数组来存储结果
        //这个任务创建并启动3个子任务
        new Task(() => results[0] = Sum(10000), TaskCreationOptions.AttachedToParent).Start();
        new Task(() => results[1] = Sum(20000), TaskCreationOptions.AttachedToParent).Start();
        new Task(() => results[2] = Sum(30000), TaskCreationOptions.AttachedToParent).Start();
        return results;
    });
    // 父任务及其子任务运行完成后，用一个延续任务显示结果
    var cwt = parent.ContinueWith(
        parentTask => Array.ForEach(parentTask.Result, Console.WriteLine));
    parent.Start();
}

注意:一个Task创建的Task默认为顶级任务，与创建它的Task无关。但是使用TaskCreationOptions.AttachedToParent标志关联Task和创建它的Task，除非所有子Task结束，否则父Task不认为已经结束。

---

任务工厂（TaskFactory和TaskFactory）

有时需要创建一组共享相同配置的Task对象，可创建一个任务工厂来封装通用的配置。System.Threading.Tasks命名空间提供了TaskFactory和TaskFactory。
构造一个任务工厂，向它传递希望任务具有的CancellationToken，TaskScheduler，TaskCreationOptions和TaskContinuationOptions设置。

public static void TaskFactoryDemo()
{
    Task parent = new Task(() =>
    {
        var cts = new CancellationTokenSource();
        var tf = new TaskFactory<Int32>(
            cts.Token,
            TaskCreationOptions.AttachedToParent,
            TaskContinuationOptions.ExecuteSynchronously,
            TaskScheduler.Default);
        var childTasks = new[]
        {
            tf.StartNew(() => Sum(cts.Token, 10000)),
            tf.StartNew(() => Sum(cts.Token, 20000)),
            tf.StartNew(() => Sum(cts.Token, Int32.MaxValue))
        };
        // 任何子任务抛出异常，就取消其余子任务
        for (int task = 0; task < childTasks.Length; task++)
        {
            childTasks[task].ContinueWith(
                t => cts.Cancel(), TaskContinuationOptions.OnlyOnFaulted);
        }
        // 所有子任务完成后，从未出错/未取消的任务获取返回的最大值
        // 然后将最大值传给另一个任务来显示最大结果。
        // **注意**：由于这个任务由任务工厂创建，所以会任务是父任务的一个子任务（默认配置）
        // 显示CancellationToken.None，和TaskContinuationOptions.ExecuteSynchronously覆盖默认值
        tf.ContinueWhenAll(
            childTasks,
            completedTasks => completedTasks.Where(
                t => !t.IsFaulted && !t.IsCanceled).Max(t => t.Result),
            CancellationToken.None)
            .ContinueWith(t => Console.WriteLine("The maxium is:" + t.Result),
                TaskContinuationOptions.ExecuteSynchronously);
    });
    // 子任务完成后，显示任何未处理的异常
    parent.ContinueWith(p =>
    {
        StringBuilder sb = new StringBuilder(
            "The following exception(s) occurred:" + Environment.NewLine);
        foreach (var e in p.Exception.Flatten().InnerExceptions)
        {
            sb.AppendLine(" " + e.GetType().ToString());
        }
        Console.WriteLine(sb.ToString());
    }, TaskContinuationOptions.OnlyOnFaulted);
    parent.Start();
}

注意：任务工厂的ContinueWhenAll和ContinueWhenAny时，NotOnRanToCompletion，NotOnFaulted和NotOnCanceled标识是非法的。组合标识当然也是非法的（OnlyOnCanceled，OnlyOnFaulted 和OnlyOnRanToCompletion）也是非法的。即，无论前置任务是如何完成的，ContinueWhenAll和ContinueWhenAny都会执行延续任务。

来自为知笔记(Wiz)