欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页  >  IT编程

C#中使用迭代器处理等待任务

程序员文章站 2023-01-21 10:04:46
 介绍 可能你已经阅读 c#5 关于 async 和 await 关键字以及它们如何帮助简化异步编程的,可惜的是在升级vs2010后短短两年时间,任然没有准备好...

 介绍

可能你已经阅读 c#5 关于 async 和 await 关键字以及它们如何帮助简化异步编程的,可惜的是在升级vs2010后短短两年时间,任然没有准备好升级到vs2012,在vs2010和c#4中不能使用异步关键字,你可能会想 “如果我能在vs 2010中写看起来同步的方法,但异步执行.我的代码会更清晰.”

看完这篇文章后,您将能够做到这一点。我们将开发一个小的基础结构代码,让我们写"看起来同步的方法,但异步执行"的方法,这个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开始):迭代器(yield return 的方式)。这里的方法是使用一个迭代器方法在c#4中建造状态机,返回一系列在全部处理过程中的等待步骤的task。我们可以编写一个方法接收一个从迭代器返回的任务的枚举,返回一个重载过的task来代表全部序列的完成以及提供它的最终结果(如果有)。

最终目标

stephen covey 建议我们目标有先后。这就是我们现在做的。已经有大量例子来告诉我们如何使用async/await来实现slams(synchronous-looking asynchronous methods)。那么我们不使用这些关键字如何实现这个功能。我们来做一个c#5 async的例子,看看如何在c#4里实现它。然后我们讨论一下转换这些代码的一般方法。

下面的例子展示了我们在c#5里实现异步读写方法stream.copytoasync()的一种写法。假设这个方法并没有在.net5里实现。
 

public static async task copytoasync(
  this stream input, stream output,
  cancellationtoken cancellationtoken = default(cancellationtoken))
{
  byte[] buffer = new byte[0x1000];  // 4 kib
  while (true) {
    cancellationtoken.throwifcancellationrequested();
    int bytesread = await input.readasync(buffer, 0, buffer.length);
    if (bytesread == 0) break;
 
    cancellationtoken.throwifcancellationrequested();
    await output.writeasync(buffer, 0, bytesread);
  }
}


对c#4,我们将分成两块:一个是相同访问能力的方法,另一个是私有方法,参数一样但返回类型不同。私有方法用迭代实现同样的处理,结果是一连串等待的任务(ienumerable<task>)。序列中的实际任务可以是非泛型或者不同类型泛型的任意组合。(幸运的是,泛型task<t>类型是非泛型task类型的子类型)

相同访问能力(公用)方法返回与相应async方法一致的类型:void,task,或者泛型task<t>。它将使用扩展方法调用私有迭代器并转化为task或者task<t>。
 

public static /*async*/ task copytoasync(
  this stream input, stream output,
  cancellationtoken cancellationtoken = default(cancellationtoken))
{
  return copytoasynctasks(input, output, cancellationtoken).totask();
}
private static ienumerable<task> copytoasynctasks(
  stream input, stream output,
  cancellationtoken cancellationtoken)
{
  byte[] buffer = new byte[0x1000];  // 4 kib
  while (true) {
    cancellationtoken.throwifcancellationrequested();
    var bytesreadtask = input.readasync(buffer, 0, buffer.length);
    yield return bytesreadtask;
    if (bytesreadtask.result == 0) break;
 
    cancellationtoken.throwifcancellationrequested();
    yield return output.writeasync(buffer, 0, bytesreadtask.result);
  }
}

异步方法通常以"async"结尾命名(除非它是事件处理器如startbutton_click)。给迭代器以同样的名字后跟“tasks”(如startbutton_clicktasks)。如果异步方法返回void值,它仍然会调用totask()但不会返回task。如果异步方法返回task<x>,那么它就会调用通用的totask<x>()扩展方法。对应三种返回类型,异步可替代的方法像下面这样:
 

public /*async*/ void dosomethingasync() {
  dosomethingasynctasks().totask();
}
public /*async*/ task dosomethingasync() {
  return dosomethingasynctasks().totask();
}
public /*async*/ task<string> dosomethingasync() {
  return dosomethingasynctasks().totask<string>();
}

成对的迭代器方法不会更复杂。当异步方法等待非通用的task时,迭代器简单的将控制权转给它。当异步方法等待task结果时,迭代器将task保存在一个变量中,转到该方法,之后再使用它的返回值。两种情况在上面的copytoasynctasks()例子里都有显示。

对包含通用resulttask<x>的slam,迭代器必须将控制转交给确切的类型。totask<x>()将最终的task转换为那种类型以便提取其结果。经常的你的迭代器将计算来自中间task的结果数值,而且仅需要将其打包在task<t>中。.net 5为此提供了一个方便的静态方法。而.net 4没有,所以我们用taskex.fromresult<t>(value)来实现它。

最后一件你需要知道的事情是如何处理中间返回的值。一个异步的方法可以从多重嵌套的块中返回;我们的迭代器简单的通过跳转到结尾来模仿它。

 

// c#5
public async task<string> dosomethingasync() {
  while (…) {
    foreach (…) {
      return "result";
    }
  }
}
 
// c#4; dosomethingasync() is necessary but omitted here.
private ienumerable<task> dosomethingasynctasks() {
  while (…) {
    foreach (…) {
      yield return taskex.fromresult("result");
      goto end;
    }
  }
end: ;
}

现在我们知道如何在c#4中写slam了,但是只有实现了fromresult<t>()和两个 totask()扩展方法才能真正的做到。下面我们开始做吧。

简单的开端

我们将在类system.threading.tasks.taskex下实现3个方法, 先从简单的那2个方法开始。fromresult()方法先创建了一个taskcompletionsource(), 然后给它的result赋值,最后返回task。
 

public static task<tresult> fromresult<tresult>(tresult resultvalue) {
  var completionsource = new taskcompletionsource<tresult>();
  completionsource.setresult(resultvalue);
  return completionsource.task;
}

很显然, 这2个totask()方法基本相同, 唯一的区别就是是否给返回对象task的result属性赋值. 通常我们不会去写2段相同的代码, 所以我们会用其中的一个方法来实现另一个。 我们经常使用泛型来作为返回结果集,那样我们不用在意返回值同时也可以避免在最后进行类型转换。 接下来我们先实现那个没有用泛型的方法。
 

private abstract class voidresult { }
 
public static task totask(this ienumerable<task> tasks) {
  return totask<voidresult>(tasks);
}

目前为止我们就剩下一个 totask<t>()方法还没有实现。

第一次天真的尝试

对于我们第一次尝试实现的方法,我们将枚举每个任务的wait()来完成,然后将最终的任务做为结果(如果合适的话)。当然,我们不想占用当前线程,我们将另一个线程来执行循环该任务。
 

// bad code !
public static task<tresult> totask<tresult>(this ienumerable<task> tasks)
{
  var tcs = new taskcompletionsource<tresult>();
  task.factory.startnew(() => {
    task last = null;
    try {
      foreach (var task in tasks) {
        last = task;
        task.wait();
      }
 
      // set the result from the last task returned, unless no result is requested.
      tcs.setresult(
        last == null || typeof(tresult) == typeof(voidresult)
          ? default(tresult) : ((task<tresult>) last).result);
 
    } catch (aggregateexception aggrex) {
      // if task.wait() threw an exception it will be wrapped in an aggregate; unwrap it.
      if (aggrex.innerexceptions.count != 1) tcs.setexception(aggrex);
      else if (aggrex.innerexception is operationcanceledexception) tcs.setcanceled();
      else tcs.setexception(aggrex.innerexception);
    } catch (operationcanceledexception cancex) {
      tcs.setcanceled();
    } catch (exception ex) {
      tcs.setexception(ex);
    }
  });
  return tcs.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,因此失去了那种可能。

是需要想点办法的时候了!

连续循环

最大的想法是直接从迭代器中获取其所产生的第一个任务。 我们创建了一个延续,使其在完成时能够检查任务的状态并且(如果成功的话)能接收下一个任务和创建另一个延续直至其结束。(如果没有,即迭代器没有需要完成的需求。)

 

// 很牛逼,但是我们还没有。
public static task<tresult> totask<tresult>(this ienumerable<task> tasks)
{
  var taskscheduler =
    synchronizationcontext.current == null
      ? taskscheduler.default : taskscheduler.fromcurrentsynchronizationcontext();
  var tcs = new taskcompletionsource<tresult>();
  var taskenumerator = tasks.getenumerator();
  if (!taskenumerator.movenext()) {
    tcs.setresult(default(tresult));
    return tcs.task;
  }
 
  taskenumerator.current.continuewith(
    t => totaskdoonestep(taskenumerator, taskscheduler, tcs, t),
    taskscheduler);
  return tcs.task;
}
private static void totaskdoonestep<tresult>(
  ienumerator<task> taskenumerator, taskscheduler taskscheduler,
  taskcompletionsource<tresult> tcs, task completedtask)
{
  var status = completedtask.status;
  if (status == taskstatus.canceled) {
    tcs.setcanceled();
 
  } else if (status == taskstatus.faulted) {
    tcs.setexception(completedtask.exception);
 
  } else if (!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>类型结束时,用一个友好的异常报错。
 

public static task<tresult> totask<tresult>(this ienumerable<task> tasks) {
  var taskscheduler =
    synchronizationcontext.current == null
      ? taskscheduler.default : taskscheduler.fromcurrentsynchronizationcontext();
  var taskenumerator = tasks.getenumerator();
  var completionsource = new taskcompletionsource<tresult>();
 
  // clean up the enumerator when the task completes.
  completionsource.task.continuewith(t => taskenumerator.dispose(), taskscheduler);
 
  totaskdoonestep(taskenumerator, taskscheduler, completionsource, null);
  return completionsource.task;
}
 
private static void totaskdoonestep<tresult>(
  ienumerator<task> taskenumerator, taskscheduler taskscheduler,
  taskcompletionsource<tresult> completionsource, task completedtask)
{
  // check status of previous nested task (if any), and stop if canceled or faulted.
  taskstatus status;
  if (completedtask == null) {
    // this is the first task from the iterator; skip status check.
  } else if ((status = completedtask.status) == taskstatus.canceled) {
    completionsource.setcanceled();
    return;
  } else if (status == taskstatus.faulted) {
    completionsource.setexception(completedtask.exception);
    return;
  }
 
  // find the next task in the iterator; handle cancellation and other exceptions.
  boolean havemore;
  try {
    havemore = taskenumerator.movenext();
 
  } catch (operationcanceledexception cancexc) {
    completionsource.setcanceled();
    return;
  } catch (exception exc) {
    completionsource.setexception(exc);
    return;
  }
 
  if (!havemore) {
    // no more tasks; set the result (if any) from the last completed task (if any).
    // we know it's not canceled or faulted because we checked at the start of this method.
    if (typeof(tresult) == typeof(voidresult)) {    // no result
      completionsource.setresult(default(tresult));
 
    } else if (!(completedtask is task<tresult>)) {   // wrong result
      completionsource.setexception(new invalidoperationexception(
        "asynchronous iterator " + taskenumerator +
          " requires a final result task of type " + typeof(task<tresult>).fullname +
          (completedtask == null ? ", but none was provided." :
            "; the actual task type was " + completedtask.gettype().fullname)));
 
    } else {
      completionsource.setresult(((task<tresult>) completedtask).result);
    }
 
  } else {
    // when the nested task completes, continue by performing this function again.
    taskenumerator.current.continuewith(
      nexttask => totaskdoonestep(taskenumerator, taskscheduler, completionsource, nexttask),
      taskscheduler);
  }
}

瞧! 现在你会在visual studio 2010中用没有async和await的 c#4 (或 vb10)写slams(看起来同步的方法,但异步执行)。

有趣的地方

直到最后那个版本,我一直在给totask()传递一个cancellationtokenup,并且将它传播进后续部分的totaskdoonestep()。(这与本文毫不相关,所以我去掉了它们。你可以在样例代码中看注释掉的痕迹。)这有两个原因。第一,处理operationcanceledexception时,我会检查它的cancellationtoken以确认它与这个操作是匹配的。如果不是,它将用一个错误来代替取消动作。虽然技术上没错,但不幸的是取消令牌可能会混淆,在其传递给totask()调用和后续部分之间的无关信息使它不值得。(如果你们这些 task专家能给我一个注释里的可确认发生的好的用例,我会重新考虑)

第二个原因是我会检查令牌是否取消,在每次movenext()调用迭代器之前,立即取消主task时,和退出进程的时候。这使你可以不经过迭代器检查令牌,具有取消的行为。我不认为这是要做的正确事情(因为对一个异步进程在yield return处取消是不合适的)——更可能是它完全在迭代器进程控制之下——但我想试试。它无法工作。我发现在某些情形,task会取消而却后续部分不会触发。请看样例代码;我靠继续执行来恢复按钮可用,但它没有发生因此按钮在进程结束之后仍不可用。我在样例代码中留下了注释掉的取消检测;你可以将取消令牌的方法参数放回去并测试它。(如果你们task专家能解释为什么会是这种情形,我将很感激!)