计算限制的异步操作(上)
一、协作式取消
.Net 对于视图取消操作的代码提供了标准的取消模式,称为协作式取消。
协作式取消:要取消的操作必须显示支持取消;
标准的协作式取消模式中两个FCL类型:
- System.Threading.CancellationTokenSource对象;
- System.Threading.CalcellationToken对象;
1.1 CancellationTokenSource
该类结构如下:
public sealed class CancellationTokenSource : IDisposable {
public CancellationTokenSource();
public void Dispose(); // 释放资源,如WaitHandle
public Boolean IsCancellationRequset { get; }
public CancellationToken Token { get; }
public void Cancel();
public void Cancel(Boolean throwOnFirstException);
}
其Token属性包含对一个或多个CancellationToken实例,并将它传递给操作。
对于使用 Token 的 Register 的多个回调,当使用 Cancel(Boolean throwOnFirstException) 方法时,对于throwOnFirstException参数:
- true:发生异常的第一个回调方法将阻止其他回调方法的调用,异常将从Cancel抛出;
- false:所有回调方法都会被调用,当所有回调执行完毕后,Cancel会抛出一个 AggregateException,该异常实例的 InnerExceptions 属性包含了所有异常对象的集合。
1.2 CancellationToken
该类型是一个结构体类型,为值类型,其常用成员如下:
public struct CancellationToken {
public static CancellationToken None {get;}
public Boolean IsCancellationRequested { get; } // 由通过非Task调用的操作调用
public void ThrowIfCancelltionRequest(); // 由通过Task调用的操作调用
public Boolean CanCanceled { get; }
// CancellationTokenSource取消时,WaitHandle会收到信号
public WaitHandle WaitHandle { get; }
public CancellationTokenRegistration Register(Action<Object> callBack,
Object state,
Boolean useSynchronizationContext);
}
该类型的实例包含一个对CancellationTokenSource对象引用的私有字段。在代码中可定时调用 IsCancellationRequested 属性判断当前操作是否需要提前退出。
通过 CancellationToken.None 这个静态属性,可返回一个不和任何 CancellationTokenSource 对象相关联的Token(该Token的对CancellationTokenSource 引用的私有字段为null)。其 CanBeCancelled 始终为false,而通过CancellationTokenSource返回的Token,该属性始终为true。
可调用 CancellationToken 的Register方法注册一个或多个将在关联的额CancellationTokenSource被取消时要调用的方法。
/// <param name="callBack">CancellationTokenSource取消时回调委托</param>
/// <param name="state">通过委托传给回调方法的状态值</param>
/// <param name="useSyncContext">是否使用调用线程的 SynchronizationContext 来调用委托</param>
public CancellationTokenRegistration Register(Action<Object> callBack,
Object state, Boolean useSyncContext);
对于参数useSynchronizationContext:
- false:使用调用Cancel的线程,顺序调用已登记的所有方法;
- true:已登记的回调方法会被 send 给已捕捉的 SynchronizationContext 对象,后者决定哪个线程调用回调;
- send:调用线程堵塞,等待在目标线程处理完毕后才会返回,相当于同步调用;
- post:将操作post到一个队列,调用线程会立即返回,相当于异步调用。
对于返回值 CancellationTokenRegisteation 对象,存在一个Dispose方法,可以清楚Token上所登记的所有回调方法。当 CancellationTokenSource 被 Cancel 时,任何回调方法都不会被调用。
1.3 CancellationTokenSource 的链接操作
// 创建两个CancellationTokenSource
var cts1 = new CancellationTokenSource();
var cts2 = new CancellationTokenSource();
cts1.Token.Register(()=> Console.WriteLine("cts1 canceled."));
cts2.Token.Regsiter(()=> Console.WriteLine("cts2 canceled."));
// 创建一个新的CancellationTokenSource,它在cts1或cts2被取消时取消
var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(cts1.Token, cts2.Token);
linkedCts.Token.Register(()=> Console.WriteLine("linkedCts canceled."));
// 取消其中一个CancellationTokenSource对象
cts2.Cancel();
// 显示哪个CancellationTokenSource对象被取消了
Console.WriteLine("cts1 canceled={0}, cts2 canceled={1}, linkedCts={2}",
cts1.IsCancellationRequested,cts2.IsCancellationRequested,linkedCts.IsCancellationRequested);
以上代码输出为:
linkedCts canceled
cts2 canceled
cts1 canceled=false, cts2.canceled=true, linkedCts=true
1.4 定时取消
可以通过以下方法构建一个定时自动取消的 CancellationTokenSource 对象,或调用 CancelAfter 方法在指定时间后自动取消:
public CancellationTokenSource(Int32 millisecondDelay);
public CancellationTokenSource(TimeSpan delay);
public void CancelAfter(Int32 millisecondDelay);
public void CancelAfter(TimeSpan delay);
二、任务
可使用两种方式创建一个Task:
- 通过构造方法,通过Action或Action<Object>来确定想要执行的操作;
- 通过静态Run方法,通过Action或Func>TResult<来确定想要执行的操作;
无论是构造器还是Run方法,都可以选择一个CancellationToken,它可以让Task能够在调度前取消。
在构造Task时,可以选择向构造器传递一些 TaskCreationOptions 标志来控制 Task 的执行方式,其定义如下:
[Flags, Serializable]
public enum TaskCreationOptions {
None = 0x0000, // Default
PreferFairness = 0x0001, // 【提议】TaskScheduler,希望任务尽快执行
LongRunning = 0x0002, // 【提议】TaskScheduler,尽可能创建线程池线程
AttachedToPrent = 0x0004, // 【设置】附加到它的父Task
DenyChildAttach = 0x0008, // 【设置】拒绝任何子任务附加
HideScheduler = 0x0010 // 【设置】使用默认TaskScheduler,而不是父任务的Scheduler
}
TaskCreationOptions控制的是任务调度器TaskScheduler对Task的操作行为,对于TaskScheduler相关的设置,TaskScheduler可能会也可能不会采纳,而后三项只和Task自身相关,总是有效的。
2.1 等待任务完成并获取结果
对于通过构造器创建完毕的Task对象,可调用Start()方法来开启任务。
可以通过调用Wait方法或Result属性来堵塞调用线程等待任务执行完毕。
Result属性内部会调用 Wait;
如果Task还没有开始执行,系统可能(取决于TaskScheduler)使用调用Wait的线程来执行Task。在这种情况下,调用Wait的线程不会堵塞;它会执行Task并立即返回,好处在于,没有线程会被堵塞,所以减少了对资源的占用(因为不需要创建一个线程来替代被堵塞的线程),并提升了性能(因为不需要花时间创建线程,也没有上下文切换)。不好的地方在于,假如线程在调用Wait前已经获得了一个线程同步锁,而Task视图获取同一个锁,就会造成死锁的线程!
以上摘自《via clr c#》中的原话,此处并不理解。
Task既可以使用实例方法 Wait 来等待一个单个任务,也可以使用静态方法等待一个Task数组:
- int Task.WaitAny(params Task[] tasks);
- 堵塞调用线程,直到数组中任一任务完成,方法即返回;
- 返回值为数组索引,若发生超时则返回-1;
- Boolean Task.WaitAll(params Task[] tasks);
- 堵塞调用线程,直到数组中所有任务完成,方法返回;
- 所有任务完成返回true,发生超时返回false;
以上两个方法若通过一个CancellationToken取消,都会抛出一个OperationCancelledException;
2.2 任务中的异常处理
默认情况下,Task代码中抛出的异常会被“吞噬”并存储到一个AggregateException对象中,线程池线程可以返回到线程池中;而当Task调用Wait系方法或Result属性来等待任务执行过程中,若任务代码发生了异常,则Wait系方法或Result属性将抛出一个System.AggregateException对象。
Task“吞噬”掉的异常可以向TaskScheduler的静态UnobervedTaskException事件等级一个回调方法,当这个Task会GC回收时,CLR的终结器线程就会引发这个事件:
TaskScheduler.UnobservedTaskException += TaskScheduler_UnobservedTaskException; private void TaskScheduler_UnobservedTaskException(object sender, UnobservedTaskExceptionEventArgs e) { UnhandledExceptionOccurred(e.Exception); e.SetObserved(); }
AggregateExceotion中包含一个InnerExceptions属性,该属性返回一个ReadOnlyCollection<Exception>集合对象(若父任务包含多个子任务,多个子任务中都抛出了异常,则集合中可能会包含多个对象)。
AggregateException提供的常用成员有:
- 重写了Exception的GetBaseException方法:
- Flatten方法:返回一个新的AggregateException,其InnerException属性是通过原始的AggregateException的内层异常层次结构而生成的。
- Handle方法:为AggregateException中包含的每个异常都调用一个回调方法,回调方法可以为每个异常决定如何对其进行处理,回调方法的返回值表示该异常已处理或未处理;
调用Handle后,如果至少存在一个未处理异常,就抛出一个新的AggregateException,其中只包含未处理的异常。
2.3 取消任务
在Task创建的时候传入一个CancellationToken,将Task和Token进行关联。若Task未启动时被取消,那么Task永远不会完成,其Status属性为 Cenceled,同时,其IsCompleted属性为true,标识该任务已经被完成。此时若再调用Start()方法来开启任务,则会抛出一个 InvalidOperationException 异常,无法开启一个已完成的方法。
CancellationTokenSource cts = new CancellationTokenSource();
Task task = new Task(() => { Sum(10000, cts.Token); }, cts.Token);
cts.Cancel();
// 这里会抛出InvalidOperationException
task.Start();
在Task启动之后想要取消任务,则必须显示支持取消,将CancellationToken作为参数传递给回调方法(或使用Lambda表达式)。
Task对象虽然关联了CancellationToken但没有办法访问它。
完整的取消Demo如下:
static void Main(string[] args)
{
CancellationTokenSource cts = new CancellationTokenSource();
Task<int> task = new Task<int>(() => Sum(10000000, cts.Token), cts.Token);
task.Start();
cts.Cancel();
try {
// 若显示取消时,任务还未完成,Result会抛出一个AggregateException
Console.WriteLine("The sum is:" + task.Result);
}
catch (AggregateException ex) {
// 将所有 OperationCanceledException 都视为已处理
// 其它任何异常在Handle中都会抛出一个AggregateException,其中只包含未处理的异常
ex.Handle(e => e is OperationCanceledException);
Console.WriteLine("Sum was canceled.");
}
Console.ReadLine();
}
public static int Sum(int n, CancellationToken token)
{
int sum = 0;
for (; n > 0; n--)
{
if (token.IsCancellationRequested) break;
checked { sum += n; };
// checked 溢出时抛出异常关键字
}
return sum;
}
2.4 任务完成时启动新任务
在任务未完成时调用Wait方法,极有可能造成线程池创建新线程。
ContinueWith用来注册当前Task执行完毕的后续任务。Task对象可多次调用ContinueWith,这样Task完成后,所有ContinueWith任务都会进入线程池队列中,使用线程池线程来完成后续任务。
- ContinueWith会返回一个新的Task对象来代表当前的任务;
- 其方法参数委托中引用了其前置Task,可以获取前置Task的执行情况。
- 可以传入TaskContinuationOptions来指定当前Task的执行条件;
TaskContinuationOptions定义如下:
[Flags, Serializable]
public enum TaskContinuationOptions {
None = 0x0000, // 默认
PreferFairness = 0x0001, // 提议TaskScheduler尽快执行任务
LongRunning = 0x0002, // 提议TaskScheduler尽可能创建线程池线程
AttachedToPrent = 0x0004, // 将当前Task和它的父Task关联
DenyChildAttach = 0x0008, // 禁止关联子任务,否则抛出InvalidOperationExcetion
HideScheduler = 0x0010, // 强迫子任务使用默调度器,而不是父任务的调度器
LazyCancellation = 0x0020, // 除非前置任务完成,否则禁止延续任务完成
ExecuteSynchronously = 0x80000, // 由执行前置任务的线程来完成当前延续任务
// 指明在什么情况下允许运行ContinueWith任务
NotOnRanToCompletion = 0x10000,
NotOnFaulted = 0x20000,
NotOnCanceled = 0x40000,
OnlyOnCanceled = NotOnRanToCompletion | NotOnFaulted,
OnlyOnFaulted = NotOnRanToCompletion | NotOnCanceled,
OnlyOnRanToCompletion = NotOnFaulted | NotOnCanceled
}
由于ContinueWith同时创建了一个新的Task,故TaskContinueationOptions也提供了TaskCreationOptions所有的选项,来设置新建的Task对象。枚举的其它选项值声明了ContinueWith任务执行的先决条件。
其中Task“完成”的各种状态如下:
- Completion:任务执行成功,没有取消也没有发生异常;
- Faulted:任务执行失败,执行任务期间发生异常;
- Canceled:任务取消,执行任务过程中被显式终止;
Task<Int32> 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);
t.ContinueWith(task => Console.WriteLine("Sum was canceled."),
TaskContinuationOptions.OnlyOnCanceled);
2.5 父子任务
任务支持父/子关系,在父任务的回调方法中可以创建新任务,并使用TaskCreationOptions将新任务指定为子任务。
只有所有子任务完成时,父任务才算结束。
Task<int[]> parentTasks = new Task<int[]>(() => {
int[] results = new int[3];
// Task<int[]> 父任务内部生成了3个子任务
Action action0 = new Action(() => results[0] = Sum(CancellationToken.None, 10000));
Action action1 = new Action(() => results[1] = Sum(CancellationToken.None, 20000));
Action action2 = new Action(() => results[2] = Sum(CancellationToken.None, 300000));
new Task(action0, TaskCreationOptions.AttachedToParent).Start();
new Task(action1, TaskCreationOptions.AttachedToParent).Start();
new Task(action2,TaskCreationOptions.AttachedToParent).Start();
return results;
});
// 所有子任务完成,父任务才算完成
parentTasks.ContinueWith(parent => {
try {
// 任何一个子任务抛出异常,父任务都不算完成,查询Result属性报错
Array.ForEach(parent.Result, Console.WriteLine);
}
catch (AggregateException ex) {
Console.WriteLine(ex.GetBaseException().Message);
}
Console.WriteLine("Parent completed status is " + parent.Status);
});
// 启动父任务
parentTasks.Start();
输出结果为
算术运算导致溢出。
Parent completed status is Faulted
2.6 任务内部揭秘
每个Task对象都有构成任务状态的一组字段(9部分):
- Id;
- Status:任务状态;
- 父任务的引用;
- 创建Task时指定的TaskScheduler的引用;
- 对回调方法的引用;
- AsyncState:对要传给回调方法的对象的引用
- ExcutionContext的引用;
- ManualResetEvenSlim对象的引用;
- 补充状态的引用:
- CancellationToken;
- ContinueWithTask对象集合;
- 抛出未处理异常的子任务准备的Task对象集合;
使用Task需要为这些状态分配内存,所以尽量使用ThreadPool.QueueUserWorkItem来节省资源;
Task和Task<T>对象都实现了IDisposable接口,默认Dispose方法都是关闭ManualResetEventSlim对象。不建议显示调用Dispose,应该让GC自己完成。
public enum TaskStatus {
//Task对象被创建
Created = 0,
//该任务正在等待 .NET Framework 基础结构在内部将其激活并进行计划。
WaitingForActivation = 1,
WaitingToRun = 2, //Task已经启动,但尚未开始执行。
Running = 3, //该任务正在运行,但尚未完成。
//该任务已完成执行,正在隐式等待附加的子任务完成。
WaitingForChildrenToComplete = 4,
RanToCompletion = 5, //已成功完成执行的任务。
Canceled = 6, //任务被取消。
Faulted = 7 //由于未处理异常的原因而完成的任务。
}
Task状态说明:
- IsCompleted只读属性:
- RanToCompletion | Canceled | Faulted 状态;
- WaitingForActivation(该状态意味着Task的调度由任务的基础结构控制):
- 通过调用 ContinueWith、ContinueWhenAll、ContinueWhenAny 方法创建的Task;
- 通过调用 FromAsync 方法创建的Task;
- 通过构造 TaskCompletionSource<TResult> 对象创建的Task;
Task出错时,其Exception属性返回一个AggregateException对象,其InnerExceptions集合包含了所有未处理的异常。
2.7 任务工厂
TaskFactory用来创建一组共享相同配置的Task对象。要向TaskFactory传递希望任务具有的 CancellationToken、TaskScheduler、TaskCreationOptions和TaskCpntinuationOptions等设置。
private static void Demo()
{
Task parent = new Task(() => {
CancellationTokenSource cts = new CancellationTokenSource();
// 所有通过TaskFactory启动的任务都是子任务,且使用父任务的线程同步执行
TaskFactory<int> tf = null;
tf = new TaskFactory<int>(cts.Token, TaskCreationOptions.AttachedToParent,
TaskContinuationOptions.ExecuteSynchronously,
TaskScheduler.Default);
// 该任务工厂创建并启动了3个子任务
Task<int>[] childTasks = new Task<int>[] {
tf.StartNew(()=> Sum(cts.Token, 10000)),
tf.StartNew(()=> Sum(cts.Token, 20000)),
tf.StartNew(()=> Sum(cts.Token, Int32.MaxValue)) // 将抛出OverFlowException
};
// 任何子任务抛出异常,就取消其余子任务
for (int t = 0; t < childTasks.Length; t++) {
TaskContinuationOptions continueOnFailed = TaskContinuationOptions.OnlyOnFaulted;
childTasks[t].ContinueWith(task => cts.Cancel(), continueOnFailed);
}
// 所有子任务完成后,从成功完成的任务中找到返回的最大值,再将该值传给另一个任务来显示最大结果
Func<Task<int>[], int> continueFunc = tasks => {
var completedTasks = tasks.Where(t => !t.IsFaulted && !t.IsCanceled);
return completedTasks.Max(item => item.Result);
};
// 当所有子任务都完成后创建一个Task,由于它是TaskFactory创建的,仍视为TaskFactory子任务
Task<int> completedTask = tf.ContinueWhenAll(childTasks, continueFunc,
CancellationToken.None);
completedTask.ContinueWith(t => Console.WriteLine("The maximum is " + t.Result),
TaskContinuationOptions.ExecuteSynchronously);
});
// 子任务完成后,也将显示任何未处理的异常
parent.ContinueWith(p => {
// 将所有文本放到一个StringBuilder中,就只用调用Console.WriteLine一次,
// 因为这个任务可能和上面的任务并行执行,而我不希望任务的输出变得不连续
string msg = "The following exception(s) occurred:" + Environment.NewLine;
StringBuilder sb = new StringBuilder(msg);
foreach (var e in p.Exception.Flatten().InnerExceptions) {
sb.AppendLine(" " + e.GetType().ToString());
}
Console.WriteLine(sb.ToString());
}, TaskContinuationOptions.OnlyOnFaulted);
// 启动父任务,使它能够启动子任务
parent.Start();
}
使用TaskFactory创建的所有任务都具有相同的配置,故ft.ContinueWhenAll返回的仍然是父任务的一个子任务,会用默认的TaskScheduler同步执行。通过向其传递CancellationToken.None来覆盖TaskFactory的CancellationToken,使其不能取消。
TaskFactory或TaskFactory<TResult>的静态ContinueWhenAll和ContinueWhenAny方法,会等待所有已创建的子任务完成后新建一个延续任务,该延续任务无论每个子任务的完成状态(Completion、Fault or Cancel)是怎样的都会执行。所以 TaskContinuationOption的以下标志是非法的:NotOnRanToCompletion, NotOnFaulted, NotOnCanceled,以及它们的组合标志。
2.8 任务调度器
TaskScheduler对象负责执行被调度的任务,同时向VS调试器公开任务信息。
FCL提供了两个派生自TaskScheduler的类型:
- 线程池任务调度器(Thread Pool Task Scheduler);
- 是所有应用程序的默认调度器,负责将任务调度给线程池的工作者线程。
- 通过
TaskScheduler.Default
获得引用;
- 同步上下文任务调度器(Synchronization Context Task Scheduler):
- 适合提供了图形用户界面的应用程序,如Winform、 WPF、SilverLight等;
- 将所有任务都调度给GUI线程,使任务代码能够成功更新UI组件;
- 该调度器不使用线程池;
- 通过
TaskScheduler.FromCurrentSynchronizationContext
方法获得引用;
注意,同步上下文调度器实际上使任务代码放到GUI线程的队列中,并没有开启新的线程,它为Task更新UI提供了一种方式。
同时,线程池线程代码不能尝试更新UI组件,否则会抛出 InvalidOperationException。
如果有特殊任务调度需求,也可以自定义TaskScheduler派生类来完成需求。
// SynchronizationContextTaskScheduler实际安排任务在主线程执行
private readonly TaskScheduler m_syncContextTaskScheduler;
public MainWindow() {
InitializeComponent();
m_syncContextTaskScheduler = TaskScheduler.FromCurrentSynchronizationContext();
txt.Text = "Synchronization Context Task Scheduler Demo";
Visibility = Visibility.Visible;
UpdateTime();
}
private CancellationTokenSource m_cts = null;
protected override void OnMouseLeftButtonDown(MouseButtonEventArgs e)
{
if (m_cts != null) { // 一个操作正在进行,取消它
m_cts.Cancel();
m_cts = null;
}
else { // 操作没有开始,启动它
txt.Text = "Operation running";
m_cts = new CancellationTokenSource();
// 这个任务使用默认TaskScheduler,在一个线程池线程上执行
Task<Int32> task = Task.Run(() => Sum(m_cts.Token, 20000), m_cts.Token);
// 这些任务使用同步上下文任务调度器,在 GUI 线程上执行
task.ContinueWith(t => txt.Text = "Result: " + t.Result, CancellationToken.None,
TaskContinuationOptions.OnlyOnRanToCompletion,
/*m_syncContextTaskScheduler*/TaskScheduler.Default);
task.ContinueWith(t => txt.Text = "Operation Canceled.", CancellationToken.None,
TaskContinuationOptions.OnlyOnCanceled,
/*m_syncContextTaskScheduler*/TaskScheduler.Default);
task.ContinueWith(t => txt.Text = "Operation Faulted.", CancellationToken.None,
TaskContinuationOptions.OnlyOnFaulted,
m_syncContextTaskScheduler);
}
base.OnMouseLeftButtonDown(e);
}
public static int Sum(CancellationToken token, int n) {
int sum = 0;
for (; n > 0; n--) {
if (token.IsCancellationRequested) break;
checked { sum += n; };
Thread.Sleep(5);
}
return sum;
}
网友评论