C# 多线程六之Task(任务)三之任务工厂
C# 多线程六之Task(任务)三之任务工厂
郑小超.
发布于 2018-12-24 10:38:03
发布于 2018-12-24 10:38:03
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.ExecuteSynchronously
tfTask.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建立关联,如果你们能找到方法,欢迎在下面评论区评论,因为这个所以,要处理子任务抛出的异常.只能通过过滤异常子任务,然后在子任务里单独记录日志的方式,去处理:
暂时没有想到更好的办法.
本文参与 腾讯云自媒体分享计划，分享自作者个人站点/博客。
原始发表：2018-12-08 ，如有侵权请联系 cloudcommunity@tencent.com 删除
评论
登录后参与评论
推荐阅读