C#中使用迭代器处理等待任务
介绍
可能你已经阅读C＃5关于async和await关键字以及它们如何帮助简化异步编程的,可惜的是在升级VS2010后短短两年时间,任然没有准备好升级到VS2012,在VS2010和C＃4中不能使用异步关键字,你可能会想“如果我能在VS2010中写看起来同步的方法,但异步执行.我的代码会更清晰.”
看完这篇文章后，您将能够做到这一点。我们将开发一个小的基础结构代码，让我们写"看起来同步的方法,但异步执行"的方法，这个VS2012异步关键字一样,享受C＃5的特性.
我们必须承认,async和await是非常好的语法糖,我们的方法需要编写更多的"AsyncResultcallback"方法适应这种变化.而当你终于升级到VS2012（或以后），这将是一件微不足道的小事，用C＃关键字替换这个方法，只要简单的语法变化，而不是一个艰苦的结构重写。
概要
async/await是基于异步任务模式的关键字。鉴于此处已经有了非常完备的文档描述，这里我就不再加以说明。但必须指出的是，TAP简直帅到极点了！通过它你可以创建大量的将在未来某时间完成的小型单元工作（任务）；任务可以启动其他的（嵌套）任务并且/或者建立一些仅当前置任务完成后才会启动的后续任务。前置与后续任务则可以链接为一对多或是多对一的关系。当内嵌任务完成时，父级任务无需与线程（重量级资源！）相绑定。执行任务时也不必再担心线程的时序安排，只需作出一些小小提示，框架将会自动为你处理这些事情。当程序开始运行，所有的任务将如溪流汇入大海般各自走向终点，又像柏青哥的小铁球一样相互反弹相互作用。
然而在C#4里面我们却没有async和await，不过缺少的也只是这一点点.Net5的新特性而已，这些新特性我们要么可以稍作回避，要么可以自己构建，关键的Task类型还是可用的。
在一个C#5的异步（async）方法里，你要等待一个Task。这不会导致线程等待；而是这个方法返回一个Task给它的调用者，这个Task能够等待（如果它自己是异步的）或者附上后续部分。（它同样能在任务中或它的结果中调用Wait()，但这会和线程耦合，所以避免那样做。）当等待的任务成功完成，你的异步方法会在它中断的地方继续运行。
也许你会知道，C#5的编译器会重写它的异步方法为一个生成的实现了状态机的嵌套类。C#正好还有一个特征（从2.0开始）：迭代器（yieldreturn的方式)。这里的方法是使用一个迭代器方法在C#4中建造状态机，返回一系列在全部处理过程中的等待步骤的Task。我们可以编写一个方法接收一个从迭代器返回的任务的枚举，返回一个重载过的Task来代表全部序列的完成以及提供它的最终结果（如果有）。
最终目标
StephenCovey建议我们目标有先后。这就是我们现在做的。已经有大量例子来告诉我们如何使用async/await来实现SLAMs(synchronous-lookingasynchronousmethods)。那么我们不使用这些关键字如何实现这个功能。我们来做一个C#5async的例子，看看如何在C#4里实现它。然后我们讨论一下转换这些代码的一般方法。
下面的例子展示了我们在C#5里实现异步读写方法Stream.CopyToAsync()的一种写法。假设这个方法并没有在.NET5里实现。
publicstaticasyncTaskCopyToAsync(
thisStreaminput,Streamoutput,
CancellationTokencancellationToken=default(CancellationToken))
{
byte[]buffer=newbyte[0x1000];//4KiB
while(true){
cancellationToken.ThrowIfCancellationRequested();
intbytesRead=awaitinput.ReadAsync(buffer,0,buffer.Length);
if(bytesRead==0)break;
cancellationToken.ThrowIfCancellationRequested();
awaitoutput.WriteAsync(buffer,0,bytesRead);
}
}
对C#4，我们将分成两块：一个是相同访问能力的方法，另一个是私有方法，参数一样但返回类型不同。私有方法用迭代实现同样的处理，结果是一连串等待的任务（IEnumerable<Task>）。序列中的实际任务可以是非泛型或者不同类型泛型的任意组合。（幸运的是，泛型Task<T>类型是非泛型Task类型的子类型）
相同访问能力（公用）方法返回与相应async方法一致的类型：void，Task，或者泛型Task<T>。它将使用扩展方法调用私有迭代器并转化为Task或者Task<T>。
publicstatic/*async*/TaskCopyToAsync(
thisStreaminput,Streamoutput,
CancellationTokencancellationToken=default(CancellationToken))
{
returnCopyToAsyncTasks(input,output,cancellationToken).ToTask();
}
privatestaticIEnumerable<Task>CopyToAsyncTasks(
Streaminput,Streamoutput,
CancellationTokencancellationToken)
{
byte[]buffer=newbyte[0x1000];//4KiB
while(true){
cancellationToken.ThrowIfCancellationRequested();
varbytesReadTask=input.ReadAsync(buffer,0,buffer.Length);
yieldreturnbytesReadTask;
if(bytesReadTask.Result==0)break;
cancellationToken.ThrowIfCancellationRequested();
yieldreturnoutput.WriteAsync(buffer,0,bytesReadTask.Result);
}
}
异步方法通常以"Async"结尾命名（除非它是事件处理器如startButton_Click）。给迭代器以同样的名字后跟“Tasks”（如startButton_ClickTasks）。如果异步方法返回void值，它仍然会调用ToTask()但不会返回Task。如果异步方法返回Task<X>，那么它就会调用通用的ToTask<X>()扩展方法。对应三种返回类型，异步可替代的方法像下面这样：
public/*async*/voidDoSomethingAsync(){
DoSomethingAsyncTasks().ToTask();
}
public/*async*/TaskDoSomethingAsync(){
returnDoSomethingAsyncTasks().ToTask();
}
public/*async*/Task<String>DoSomethingAsync(){
returnDoSomethingAsyncTasks().ToTask<String>();
}
成对的迭代器方法不会更复杂。当异步方法等待非通用的Task时，迭代器简单的将控制权转给它。当异步方法等待task结果时，迭代器将task保存在一个变量中，转到该方法，之后再使用它的返回值。两种情况在上面的CopyToAsyncTasks()例子里都有显示。
对包含通用resultTask<X>的SLAM，迭代器必须将控制转交给确切的类型。ToTask<X>()将最终的task转换为那种类型以便提取其结果。经常的你的迭代器将计算来自中间task的结果数值，而且仅需要将其打包在Task<T>中。.NET5为此提供了一个方便的静态方法。而.NET4没有，所以我们用TaskEx.FromResult<T>(value)来实现它。
最后一件你需要知道的事情是如何处理中间返回的值。一个异步的方法可以从多重嵌套的块中返回；我们的迭代器简单的通过跳转到结尾来模仿它。
//C#5
publicasyncTask<String>DoSomethingAsync(){
while(…){
foreach(…){
return"Result";
}
}
}
//C#4;DoSomethingAsync()isnecessarybutomittedhere.
privateIEnumerable<Task>DoSomethingAsyncTasks(){
while(…){
foreach(…){
yieldreturnTaskEx.FromResult("Result");
gotoEND;
}
}
END:;
}
现在我们知道如何在C#4中写SLAM了，但是只有实现了FromResult<T>()和两个ToTask()扩展方法才能真正的做到。下面我们开始做吧。
简单的开端
我们将在类System.Threading.Tasks.TaskEx下实现3个方法,先从简单的那2个方法开始。FromResult()方法先创建了一个TaskCompletionSource(),然后给它的result赋值，最后返回Task。
publicstaticTask<TResult>FromResult<TResult>(TResultresultValue){
varcompletionSource=newTaskCompletionSource<TResult>();
completionSource.SetResult(resultValue);
returncompletionSource.Task;
}
很显然,这2个ToTask()方法基本相同,唯一的区别就是是否给返回对象Task的Result属性赋值.通常我们不会去写2段相同的代码,所以我们会用其中的一个方法来实现另一个。我们经常使用泛型来作为返回结果集，那样我们不用在意返回值同时也可以避免在最后进行类型转换。接下来我们先实现那个没有用泛型的方法。
privateabstractclassVoidResult{}
publicstaticTaskToTask(thisIEnumerable<Task>tasks){
returnToTask<VoidResult>(tasks);
}
目前为止我们就剩下一个ToTask<T>()方法还没有实现。
第一次天真的尝试
对于我们第一次尝试实现的方法,我们将枚举每个任务的Wait()来完成,然后将最终的任务做为结果(如果合适的话)。当然,我们不想占用当前线程,我们将另一个线程来执行循环该任务。
//BADCODE!
publicstaticTask<TResult>ToTask<TResult>(thisIEnumerable<Task>tasks)
{
vartcs=newTaskCompletionSource<TResult>();
Task.Factory.StartNew(()=>{
Tasklast=null;
try{
foreach(vartaskintasks){
last=task;
task.Wait();
}
//Settheresultfromthelasttaskreturned,unlessnoresultisrequested.
tcs.SetResult(
last==null||typeof(TResult)==typeof(VoidResult)
?default(TResult):((Task<TResult>)last).Result);
}catch(AggregateExceptionaggrEx){
//Iftask.Wait()threwanexceptionitwillbewrappedinanAggregate;unwrapit.
if(aggrEx.InnerExceptions.Count!=1)tcs.SetException(aggrEx);
elseif(aggrEx.InnerExceptionisOperationCanceledException)tcs.SetCanceled();
elsetcs.SetException(aggrEx.InnerException);
}catch(OperationCanceledExceptioncancEx){
tcs.SetCanceled();
}catch(Exceptionex){
tcs.SetException(ex);
}
});
returntcs.Task;
}
这里有一些好东西，事实上它真的有用,只要不触及用户界面：
它准确的返回了一个TaskCompletionSource的Task，并且通过源代码设置了完成状态。
- 它显示了我们怎么通过迭代器的最后一个任务设置task的最终Result，同时避免可能没有结果的情况。
- 它从迭代器中捕获异常并设置Canceled或Faulted状态.它也传播枚举的task状态(这里是通过Wait()，该方法可能抛出一个包装了cancellation或fault的异常的集合).
但这里有些主要的问题。最严重的是：
- 由于迭代器需要实现“异步态的”的诺言，当它从一个UI线程初始化以后，迭代器的方法将能访问UI控件。你能发现这里的foreach循环都是运行在后台；从那个时刻开始不要触摸UI！这种方法没有顾及SynchronizationContext。
- 在UI之外我们也有麻烦。我们可能想制造大量大量的由SLAM实现的并行运行的Tasks。但是看看循环中的Wait()！当等待一个嵌套task时，可能远程需要一个很长的时间完成，我们会挂起一个线程。我们面临线程池的线程资源枯竭的情况。
- 这种解包Aggregate异常的方法是不太自然的。我们需要捕获并传播它的完成状态而不抛出异常。
- 有时SLAM可以立刻决定它的完成状态。那种情形下，C#5的async可以异步并且有效的操作。这里我们总是计划了一个后台task，因此失去了那种可能。
是需要想点办法的时候了！
连续循环
最大的想法是直接从迭代器中获取其所产生的第一个任务。我们创建了一个延续，使其在完成时能够检查任务的状态并且（如果成功的话）能接收下一个任务和创建另一个延续直至其结束。（如果没有，即迭代器没有需要完成的需求。）
//很牛逼，但是我们还没有。
publicstaticTask<TResult>ToTask<TResult>(thisIEnumerable<Task>tasks)
{
vartaskScheduler=
SynchronizationContext.Current==null
?TaskScheduler.Default:TaskScheduler.FromCurrentSynchronizationContext();
vartcs=newTaskCompletionSource<TResult>();
vartaskEnumerator=tasks.GetEnumerator();
if(!taskEnumerator.MoveNext()){
tcs.SetResult(default(TResult));
returntcs.Task;
}
taskEnumerator.Current.ContinueWith(
t=>ToTaskDoOneStep(taskEnumerator,taskScheduler,tcs,t),
taskScheduler);
returntcs.Task;
}
privatestaticvoidToTaskDoOneStep<TResult>(
IEnumerator<Task>taskEnumerator,TaskSchedulertaskScheduler,
TaskCompletionSource<TResult>tcs,TaskcompletedTask)
{
varstatus=completedTask.Status;
if(status==TaskStatus.Canceled){
tcs.SetCanceled();
}elseif(status==TaskStatus.Faulted){
tcs.SetException(completedTask.Exception);
}elseif(!taskEnumerator.MoveNext()){
//设置最后任务返回的结果，直至无需结果为止。
tcs.SetResult(
typeof(TResult)==typeof(VoidResult)
?default(TResult):((Task<TResult>)completedTask).Result);
}else{
taskEnumerator.Current.ContinueWith(
t=>ToTaskDoOneStep(taskEnumerator,taskScheduler,tcs,t),
taskScheduler);
}
}
这里有许多值得分享的：
我们的后续部分(continuations)使用涉及SynchronizationContext的TaskScheduler，如果有的话。这使得我们的迭代器在UI线程初始化以后，立刻或者在一个继续点被调用，去访问UI控件。
进程不中断的运行，因此没有线程挂起等待！顺便说一下，在ToTaskDoOneStep()中对自身的调用不是递归调用；它是在taskEnumerator.Currenttask结束后调用的匿名函数，当前活动在调用ContinueWith()几乎立刻退出，它完全独立于后续部分。
此外，我们在继续点中验证每个嵌套task的状态，不是检查一个预测值。
然而，这儿至少有一个大问题和一些小一点的问题。
如果迭代器抛出一个未处理异常，或者抛出OperationCanceledException而取消，我们没有处理它或设置主task的状态。这是我们以前曾经做过的但在此版本丢失了。
为了修复问题1，我们不得不在两个方法中调用MoveNext()的地方引入同样的异常处理机制。即使是现在，两个方法中都有一样的后续部分建立。我们违背了“不要重复你自己”的信条。
如果异步方法被期望给出一个结果，但是迭代器没有提供就退出了会怎么样呢？或者它最后的task是错误的类型呢？第一种情形下，我们默默返回默认的结果类型；第二种情形，我们抛出一个未处理的InvalidCastException，主task永远不会到达结束状态！我们的程序将永久的挂起。
最后，如果一个嵌套的task取消或者发生错误呢？我们设置主task状态，再也不会调用迭代器。可能是在一个using块，或带有finally的try块的内部，并且有一些清理要做。我们应当遵守过程在中断的时候使它结束，而不要等垃圾收集器去做这些。我们怎么做到呢？当然通过一个后续部分！
为了解决这些问题，我们从ToTask()中移走MoveNext()调用，取而代之一个对ToTaskDoOneStep()的初始化的同步调用。然后我们将在一个提防增加合适的异常处理。
最终版本
这里是ToTask<T>()的最终实现.它用一个TaskCompletionSource返回主task，永远不会引起线程等待，如果有的话还会涉及SynchronizationContext，由迭代器处理异常，直接传播嵌套task的结束（而不是AggregateException）,合适的时候向主task返回一个值，当期望一个结果而SLAM迭代器没有以正确的genericTask<T>类型结束时，用一个友好的异常报错。
publicstaticTask<TResult>ToTask<TResult>(thisIEnumerable<Task>tasks){
vartaskScheduler=
SynchronizationContext.Current==null
?TaskScheduler.Default:TaskScheduler.FromCurrentSynchronizationContext();
vartaskEnumerator=tasks.GetEnumerator();
varcompletionSource=newTaskCompletionSource<TResult>();
//Cleanuptheenumeratorwhenthetaskcompletes.
completionSource.Task.ContinueWith(t=>taskEnumerator.Dispose(),taskScheduler);
ToTaskDoOneStep(taskEnumerator,taskScheduler,completionSource,null);
returncompletionSource.Task;
}
privatestaticvoidToTaskDoOneStep<TResult>(
IEnumerator<Task>taskEnumerator,TaskSchedulertaskScheduler,
TaskCompletionSource<TResult>completionSource,TaskcompletedTask)
{
//Checkstatusofpreviousnestedtask(ifany),andstopifCanceledorFaulted.
TaskStatusstatus;
if(completedTask==null){
//Thisisthefirsttaskfromtheiterator;skipstatuscheck.
}elseif((status=completedTask.Status)==TaskStatus.Canceled){
completionSource.SetCanceled();
return;
}elseif(status==TaskStatus.Faulted){
completionSource.SetException(completedTask.Exception);
return;
}
//FindthenextTaskintheiterator;handlecancellationandotherexceptions.
BooleanhaveMore;
try{
haveMore=taskEnumerator.MoveNext();
}catch(OperationCanceledExceptioncancExc){
completionSource.SetCanceled();
return;
}catch(Exceptionexc){
completionSource.SetException(exc);
return;
}
if(!haveMore){
//Nomoretasks;settheresult(ifany)fromthelastcompletedtask(ifany).
//Weknowit'snotCanceledorFaultedbecausewecheckedatthestartofthismethod.
if(typeof(TResult)==typeof(VoidResult)){//Noresult
completionSource.SetResult(default(TResult));
}elseif(!(completedTaskisTask<TResult>)){//Wrongresult
completionSource.SetException(newInvalidOperationException(
"Asynchronousiterator"+taskEnumerator+
"requiresafinalresulttaskoftype"+typeof(Task<TResult>).FullName+
(completedTask==null?",butnonewasprovided.":
";theactualtasktypewas"+completedTask.GetType().FullName)));
}else{
completionSource.SetResult(((Task<TResult>)completedTask).Result);
}
}else{
//Whenthenestedtaskcompletes,continuebyperformingthisfunctionagain.
taskEnumerator.Current.ContinueWith(
nextTask=>ToTaskDoOneStep(taskEnumerator,taskScheduler,completionSource,nextTask),
taskScheduler);
}
}
瞧!现在你会在VisualStudio2010中用没有async和await的C#4(或VB10)写SLAMs(看起来同步的方法,但异步执行)。
有趣的地方
直到最后那个版本，我一直在给ToTask()传递一个CancellationTokenUp，并且将它传播进后续部分的ToTaskDoOneStep()。（这与本文毫不相关，所以我去掉了它们。你可以在样例代码中看注释掉的痕迹。）这有两个原因。第一，处理OperationCanceledException时，我会检查它的CancellationToken以确认它与这个操作是匹配的。如果不是，它将用一个错误来代替取消动作。虽然技术上没错，但不幸的是取消令牌可能会混淆，在其传递给ToTask()调用和后续部分之间的无关信息使它不值得。（如果你们这些Task专家能给我一个注释里的可确认发生的好的用例，我会重新考虑）
第二个原因是我会检查令牌是否取消，在每次MoveNext()调用迭代器之前,立即取消主task时，和退出进程的时候。这使你可以不经过迭代器检查令牌，具有取消的行为。我不认为这是要做的正确事情（因为对一个异步进程在yieldreturn处取消是不合适的）——更可能是它完全在迭代器进程控制之下——但我想试试。它无法工作。我发现在某些情形，task会取消而却后续部分不会触发。请看样例代码；我靠继续执行来恢复按钮可用，但它没有发生因此按钮在进程结束之后仍不可用。我在样例代码中留下了注释掉的取消检测；你可以将取消令牌的方法参数放回去并测试它。（如果你们Task专家能解释为什么会是这种情形，我将很感激！）