DIV CSS 佈局教程網

 DIV+CSS佈局教程網 >> 網頁腳本 >> JavaScript入門知識 >> 關於JavaScript >> C#中使用迭代器處理等待任務
C#中使用迭代器處理等待任務
編輯:關於JavaScript     

 介紹

可能你已經閱讀 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專家能解釋為什麼會是這種情形,我將很感激!)

XML學習教程| jQuery入門知識| AJAX入門| Dreamweaver教程| Fireworks入門知識| SEO技巧| SEO優化集錦|
Copyright © DIV+CSS佈局教程網 All Rights Reserved