C# 多线程六之Task(任务)三之任务工厂

1、知识回顾,简要概述

前面两篇关于Task的随笔,C# 多线程五之Task(任务)一 和 C# 多线程六之Task(任务)二,介绍了关于Task的一些基本的用法,以及一些使用的要点,如果都看懂了,本文将介绍另一个Task的特殊用法,前面介绍了,如何通过一个父任务创建多个子任务,且这些子任务都必须要支持取消的例子,常规做法是,通过new 一个Task数组对象,然后在该对象的内部创建多个Task任务,然后给这些任务指定TaskCreationOptions.AttachedToParent,这样所有的子任务都关联到了父任务,接着给这些子任务,绑定一个CancellationToken类实例,当其中一个子任务发生异常时,调用CancellationToken类实例的Cancel方法,将其余的子任务全都取消,大致代码如下:

static void Main(string[] args){var parentTask = new Task<int[]>(() =>{var results = new int[3];var cancelTokenSource = new CancellationTokenSource();var childTasks = new Task[] {new Task(() => results[0] = ChildThreadOne(cancelTokenSource.Token),cancelTokenSource.Token, TaskCreationOptions.AttachedToParent),new Task(() => results[1] = ChildThreadTwo(cancelTokenSource.Token),cancelTokenSource.Token, TaskCreationOptions.AttachedToParent),new Task(() => results[2] = ChildThreadThree(cancelTokenSource.Token),cancelTokenSource.Token, TaskCreationOptions.AttachedToParent),};//开启所有的子任务childTasks.ForEach(f => { f.Start(); });//如果有子任务发生异常,那么通过取消信号量终止所有的任务childTasks.ForEach(f =>{f.ContinueWith(task=> cancelTokenSource.Cancel(), TaskContinuationOptions.OnlyOnFaulted);});return results;});parentTask.Start();parentTask.ContinueWith(x =>{Console.WriteLine("当父任务执行完毕时,CLR会唤起一个新线程,将父任务的返回值(子任务的返回值)输出,所以这里不会有任何的线程发生阻塞");foreach (var re in parentTask.Result){Console.WriteLine("子任务的返回值分别为:{0}", re);}});Console.WriteLine("主线程不会阻塞,它会继续执行");Console.ReadKey();//必须加这行代码,因为Task时线程池线程,属于后台线程 }/// <summary>/// 子任务一/// </summary>static int ChildThreadOne(CancellationToken token){Thread.Sleep(3000);//模拟长时间计算操作 token.ThrowIfCancellationRequested();Console.WriteLine("子任务一完成了计算任务,并返回值:{0}", 6);return 6;}/// <summary>/// 子任务二/// </summary>static int ChildThreadTwo(CancellationToken token){Thread.Sleep(2000);//模拟长时间计算操作 token.ThrowIfCancellationRequested();throw new Exception("模拟抛出异常");}/// <summary>/// 子任务三/// </summary>static int ChildThreadThree(CancellationToken token){Thread.Sleep(3000);//模拟长时间计算操作 token.ThrowIfCancellationRequested();Console.WriteLine("子任务三完成了计算任务,并返回值:{0}", 6);return 6;}}/// <summary>/// Linq扩展/// </summary>public static class LinqExtension{public static void ForEach<T>(this IEnumerable<T> enumerators, Action<T> action){foreach (var item in enumerators){action(item);}}}

这里需要注意,这里给父任务parentTask开启了三个子任务,并且通过TaskCreationOptions.AttachedToParent指定了所有的子任务不能独立于父任务运行,并且给所有的子任务,传递了CancellationToken信号量,当其中一个子任务发生异常时,所有其余的子任务都终止,但是你必须知道的是,你没有判断哪个任务会被终止,因为如果不指定线程优先级,哪怕制定了优先级,你也无法确定的判断某个计算任务在什么时候会调度完,所以我給正常的执行的任务,Sleep了三秒,抛出异常的任务Sleep了两秒,所以所有的子线程都无法执行完毕.

 

2、代码重构

 ok,虽然上面的代码很好的完成了我们在代码层面的需求,但是处于对代码的重用性考虑,有没有发现这个问题:

这块操作,可以重构的,因为所有的参数都一样,当然你可以去抽象一个共有的方法,里面放一个Func委托,当然把参数抽象出来,形成一个公共的方法,像下面这样做:

class Program{private static CancellationTokenSource cancelTokenSource = new CancellationTokenSource();private static TaskCreationOptions taskCreationOptions = TaskCreationOptions.AttachedToParent;static void Main(string[] args){var parentTask = new Task<int[]>(() =>{var results = new int[3];var childTasks = new Task[] {ExecuteChildThread(task=> results[0]=ChildThreadOne(cancelTokenSource.Token)),ExecuteChildThread(task=> results[1]=ChildThreadTwo(cancelTokenSource.Token)),ExecuteChildThread(task=> results[2]=ChildThreadThree(cancelTokenSource.Token))};//开启所有的子任务childTasks.ForEach(f => { f.Start(); });//如果有子任务发生异常,那么通过取消信号量终止所有的任务childTasks.ForEach(f =>{f.ContinueWith(task=> cancelTokenSource.Cancel(), TaskContinuationOptions.OnlyOnFaulted);});return results;});parentTask.Start();parentTask.ContinueWith(x =>{Console.WriteLine("当父任务执行完毕时,CLR会唤起一个新线程,将父任务的返回值(子任务的返回值)输出,所以这里不会有任何的线程发生阻塞");foreach (var re in parentTask.Result){Console.WriteLine("子任务的返回值分别为:{0}", re);}});Console.WriteLine("主线程不会阻塞,它会继续执行");Console.ReadKey();//必须加这行代码,因为Task时线程池线程,属于后台线程 }/// <summary>/// 子任务一/// </summary>static int ChildThreadOne(CancellationToken token){Thread.Sleep(2000);//模拟长时间计算操作 token.ThrowIfCancellationRequested();Console.WriteLine("子任务一完成了计算任务,并返回值:{0}", 6);return 6;}/// <summary>/// 子任务二/// </summary>static int ChildThreadTwo(CancellationToken token){Thread.Sleep(2000);//模拟长时间计算操作 token.ThrowIfCancellationRequested();Console.WriteLine("子任务二完成了计算任务,并返回值:{0}", 6);return 6;}/// <summary>/// 子任务三/// </summary>static int ChildThreadThree(CancellationToken token){Thread.Sleep(2000);//模拟长时间计算操作 token.ThrowIfCancellationRequested();Console.WriteLine("子任务三完成了计算任务,并返回值:{0}", 6);return 6;}/// <summary>/// 创建一个通用的子线程方法,里面封装了所有子线程的需要设置的公共参数/// </summary>/// <param name="func"></param>/// <returns></returns>static Task<int> ExecuteChildThread(Func<CancellationToken, int> func){var t=new Task<int>(()=>func.Invoke(cancelTokenSource.Token), cancelTokenSource.Token, taskCreationOptions);return t;}}/// <summary>/// Linq扩展/// </summary>public static class LinqExtension{public static void ForEach<T>(this IEnumerable<T> enumerators, Action<T> action){foreach (var item in enumerators){action(item);}}}

ok,通过对子任务的抽象,你可以这么干,但是MS提供了更好的办法,你又何必重复造轮子呢?而且这里存在着潜在的多线程争用问题,

所有的线程都用到了这两个全局变量,最好加个锁,但是加了锁之后,性能就会受到影响.

但是奇怪的是,我无法重现,如果你能重现那是最好的,下面就开始介绍Ms提供的任务工厂

 

3、任务工厂实战

下面再次对上面的方法进行重构,用任务工厂的方式,首先使用TaskFactory任务工厂的前提你必须清楚,就是创建的子任务,必须是一组共享配置的子任务对象集,所以,如果当中如果某个子任务需要使用特殊的配置,那就不能使用任务工厂,也不是不能使用,就是那个子任务你必须独立出来,不能放到任务工厂里面.ok,了解了前提条件后,开始实践,代码如下:

class Program{static void Main(string[] args){var parentTask=Task.Run(()=> {var cts = new CancellationTokenSource();//通过TaskFactory设置子任务的公共参数var tf = new TaskFactory<int>(cts.Token,TaskCreationOptions.AttachedToParent,TaskContinuationOptions.ExecuteSynchronously,TaskScheduler.Default);//通过TaskFactory设置所有的子任务,这些子任务共享上面公共参数var childTasks = new Task<int>[] {tf.StartNew(() => ChildThreadOne(cts.Token)),tf.StartNew(() => ChildThreadTwo(cts.Token)),tf.StartNew(() => ChildThreadThree(cts.Token))};//如果子任务发生异常,则向余下没有执行完毕的子任务传递取消执行的信号,如果有子任务执行完毕了,那就没有办法了childTasks.ForEach(f =>{f.ContinueWith(childTask => cts.Cancel(), TaskContinuationOptions.OnlyOnFaulted);});//遍历所有通过TaskFactory创建的子任务,然后筛选出没有被取消和没有发生异常的子任务,或者这些任务中的最大返回值//这个任务不阻塞线程,只有当所有的子任务执行完毕之后,CLR会唤起线程池中的一个新线程来执行这个操作//通过给唤起子线程设置CancellationToken.None,来达到这个线程不会被任何因素来取消该线程的目的var tfTask = tf.ContinueWhenAll(childTasks,completedTasks => completedTasks.Where(completedTask => !completedTask.IsCanceled && !completedTask.IsFaulted).Max(completedTask => completedTask.Result), CancellationToken.None);//输出所有符合要求的子任务集合的返回值集合中的最大值,并指定该任务,在tfTask任务的基础上同步执行的效果通过TaskContinuationOptions.ExecuteSynchronouslytfTask.ContinueWith(childTasksCompleteTask =>{Console.WriteLine("The Max Return Value is {0}", childTasksCompleteTask.Result);},TaskContinuationOptions.ExecuteSynchronously);});Console.WriteLine("主线程继续做它的事情");Console.ReadKey();//必须加这行代码,因为Task时线程池线程,属于后台线程 }/// <summary>/// 子任务一/// </summary>static int ChildThreadOne(CancellationToken token){var returnValue = 6;Thread.Sleep(3000);//模拟长时间计算操作 token.ThrowIfCancellationRequested();Console.WriteLine("子任务一完成了计算任务,并返回值:{0}", returnValue);return returnValue;}/// <summary>/// 子任务二/// </summary>static int ChildThreadTwo(CancellationToken token){var returnValue = 66;Thread.Sleep(3000);//模拟长时间计算操作 token.ThrowIfCancellationRequested();Console.WriteLine("子任务二完成了计算任务,并返回值:{0}", returnValue);return returnValue;}/// <summary>/// 子任务三/// </summary>static int ChildThreadThree(CancellationToken token){Thread.Sleep(2000);//模拟长时间计算操作throw new Exception("模拟抛出异常");}}/// <summary>/// Linq扩展/// </summary>public static class LinqExtension{public static void ForEach<T>(this IEnumerable<T> enumerators, Action<T> action){foreach (var item in enumerators){action(item);}}}

因为我给异常线程设置了2秒的休眠时间,正常子线程设置了3秒的休眠时间,所以所有的线程都没有执行完毕,就被取消掉了.如果修改下正常线程的休眠时间为1秒,将会得到以下的输出:

so,TaskFactory完美的完成了它的任务,且不会有任务线程发生阻塞的情况。

 

4、如何解决任务工厂抛出的异常

我发现一个很奇怪的问题,就是当当外部通过一个Task.Run创建的父任务,无法获取TaskFactory下子任务集群抛出的异常,代码如下:

class Program{static void Main(string[] args){var pTask = Task.Run(() =>{var cts = new CancellationTokenSource();//通过TaskFactory设置子任务的公共参数var tf = new TaskFactory<int>(cts.Token, TaskCreationOptions.AttachedToParent, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default);//通过TaskFactory设置所有的子任务,这些子任务共享上面公共参数var childTasks = new Task<int>[] {tf.StartNew(() => ChildThreadOne(cts.Token)),tf.StartNew(() => ChildThreadTwo(cts.Token)),tf.StartNew(() => ChildThreadThree(cts.Token))};});pTask.ContinueWith(tasks =>{var exceptions = tasks.Exception;foreach (var ex in exceptions.InnerExceptions){Console.WriteLine(ex);}},TaskContinuationOptions.OnlyOnFaulted);Console.WriteLine("主线程继续做它的事情");Console.ReadKey();//必须加这行代码,因为Task时线程池线程,属于后台线程 }/// <summary>/// 子任务一/// </summary>static int ChildThreadOne(CancellationToken token){var returnValue = 6;Thread.Sleep(3000);//模拟长时间计算操作 token.ThrowIfCancellationRequested();Console.WriteLine("子任务一完成了计算任务,并返回值:{0}", returnValue);return returnValue;}/// <summary>/// 子任务二/// </summary>static int ChildThreadTwo(CancellationToken token){var returnValue = 66;Thread.Sleep(3000);//模拟长时间计算操作 token.ThrowIfCancellationRequested();Console.WriteLine("子任务二完成了计算任务,并返回值:{0}", returnValue);return returnValue;}/// <summary>/// 子任务三/// </summary>static int ChildThreadThree(CancellationToken token){Thread.Sleep(2000);//模拟长时间计算操作throw new Exception("模拟抛出异常");}}/// <summary>/// Linq扩展/// </summary>public static class LinqExtension{public static void ForEach<T>(this IEnumerable<T> enumerators, Action<T> action){foreach (var item in enumerators){action(item);}}}

很其怪,不过这说明,外部的父任务,无法和TaskFactory建立关联,如果你们能找到方法,欢迎在下面评论区评论,因为这个所以,要处理子任务抛出的异常.只能通过过滤异常子任务,然后在子任务里单独记录日志的方式,去处理:

暂时没有想到更好的办法.

 

转载于:https://www.cnblogs.com/GreenLeaves/p/10088635.html

总结

以上是生活随笔为你收集整理的C# 多线程六之Task(任务)三之任务工厂的全部内容，希望文章能够帮你解决所遇到的问题。

如果觉得生活随笔网站内容还不错，欢迎将生活随笔推荐给好友。