TPL 和傳統 .NET 非同步程式設計
.NET Framework 提供下列兩個標準模式來執行 I/O 繫結和計算繫結的非同步作業：
非同步程式設計模型 (APM)，在這個模式中非同步作業會以一對 Begin/End 方法 (例如 FileStream.BeginRead 和 Stream.EndRead) 表示。
事件架構非同步模式 (EAP)，在這個模式中非同步作業會以一對名為 <作業名稱>Async/<作業名稱>Completed 的方法/事件 (例如 WebClient.DownloadStringAsync 和 WebClient.DownloadStringCompleted) 表示 (EAP 是在 .NET Framework 2.0 版中首次引進)。
工作平行程式庫 (TPL) 可用各種方式搭配任一種非同步模式一起使用。 您可以將 APM 和 EAP 作業以 Task 的形式公開給程式庫消費者，也可以公開 APM 模式，但是在內部使用 Task 物件來實作 APM 模式。 在這兩種情況下，您可以使用 Task 物件來簡化程式碼並利用下列實用的功能：
在工作開始之後，隨時以接續工作的形式來註冊回呼。
使用 ContinueWhenAll 和 ContinueWhenAny 方法，或是 WaitAll 方法或 WaitAny 方法，協調多個為回應 Begin_ 方法而執行的作業。
將非同步的 I/O 繫結作業和計算繫結作業封裝在同一個 Task 物件中。
監視 Task 物件的狀態。
使用 TaskCompletionSource<TResult> 將作業的狀態封送處理至 Task 物件。
將 APM 作業包裝在工作中
System.Threading.Tasks.TaskFactory 和 System.Threading.Tasks.TaskFactory<TResult> 類別都提供 FromAsync 和 FromAsync 方法的數個多載，這些多載可讓您將一對 APM Begin/End 方法封裝在一個 Task 執行個體或 Task<TResult> 執行個體中。 這些多載可容納任何具有零到三個輸入參數的 Begin/End 方法配對。
如果配對中的 End 方法會傳回值 (在 Visual Basic 中為 Function)，請使用 TaskFactory<TResult> (此會建立 Task<TResult>) 中的方法。 如果 End 方法會傳回 void (在 Visual Basic 中為 Sub)，請使用 TaskFactory (此會建立 Task) 中的方法。
在 Begin 方法具有多於三個參數或是包含 ref 或 out 參數的少數案例中，則會提供其他只封裝 End 方法的 FromAsync 多載。
下列程式碼範例顯示 FromAsync 多載的簽章，這個簽章符合 FileStream.BeginRead 和 FileStream.EndRead 方法。 這個多載會接受如下所示的三個輸入參數。
Public Function FromAsync(Of TArg1, TArg2, TArg3)(
ByVal beginMethod As Func(Of TArg1, TArg2, TArg3, AsyncCallback, Object, IAsyncResult),
ByVal endMethod As Func(Of IAsyncResult, TResult),
ByVal dataBuffer As TArg1,
ByVal byteOffsetToStartAt As TArg2,
ByVal maxBytesToRead As TArg3,
ByVal stateInfo As Object)
public Task<TResult> FromAsync<TArg1, TArg2, TArg3>(
Func<TArg1, TArg2, TArg3, AsyncCallback, object, IAsyncResult> beginMethod, //BeginRead
Func<IAsyncResult, TResult> endMethod, //EndRead
TArg1 arg1, // the byte[] buffer
TArg2 arg2, // the offset in arg1 at which to start writing data
TArg3 arg3, // the maximum number of bytes to read
object state // optional state information
)
第一個參數是 Func<T1, T2, T3, T4, T5, TResult> 委派，這個委派符合 FileStream.BeginRead 方法的簽章。 第二個參數是 Func<T, TResult> 委派，這個委派會接受 IAsyncResult 並傳回 TResult。 因為 EndRead 會傳回整數，所以編譯器會推斷 TResult 的型別為 Int32，並推斷工作的型別為 Task<Int32>。 最後四個參數與 FileStream.BeginRead 方法中的參數相同：
用以儲存檔案資料的緩衝區。
緩衝區中要開始讓資料寫入的位移位置。
可從檔案讀取的最大資料量。
選擇性物件，用以儲存要傳遞至回呼的使用者定義狀態資料。
使用 ContinueWith 回呼功能
如果您需要存取檔案中的資料，而不只是想知道位元組數目，FromAsync 方法並不足夠。 請改用 Task<String>，其 Result 屬性會包含檔案資料。 在作法上，您只要將接續工作加入至原始工作即可。 接續工作會執行通常由 AsyncCallback 委派在做的事。 當前項工作完成且資料緩衝區已填滿，就會叫用接續工作 (在返回之前 FileStream 物件應要已關閉)。
下列範例顯示如何傳回 Task<String>，這個方法會封裝 FileStream 類別的 BeginRead/EndRead 配對。
Const MAX_FILE_SIZE As Integer = 14000000
Shared Function GetFileStringAsync(ByVal path As String) As Task(Of String)
Dim fi As New FileInfo(path)
Dim data(fi.Length) As Byte
Dim fs As FileStream = New FileStream(path, FileMode.Open, FileAccess.Read, FileShare.Read, data.Length, True)
' Task(Of Integer) returns the number of bytes read
Dim myTask As Task(Of Integer) = Task(Of Integer).Factory.FromAsync(
AddressOf fs.BeginRead, AddressOf fs.EndRead, data, 0, data.Length, Nothing)
' It is possible to do other work here while waiting
' for the antecedent task to complete.
' ...
' Add the continuation, which returns a Task<string>.
Return myTask.ContinueWith(Function(antecedent)
fs.Close()
If (antecedent.Result < 100) Then
Return "Data is too small to bother with."
End If
' If we did not receive the entire file, the end of the
' data buffer will contain garbage.
If (antecedent.Result < data.Length) Then
Array.Resize(data, antecedent.Result)
End If
' Will be returned in the Result property of the Task<string>
' at some future point after the asynchronous file I/O operation completes.
Return New UTF8Encoding().GetString(data)
End Function)
End Function
const int MAX_FILE_SIZE = 14000000;
public static Task<string> GetFileStringAsync(string path)
{
FileInfo fi = new FileInfo(path);
byte[] data = null;
data = new byte[fi.Length];
FileStream fs = new FileStream(path, FileMode.Open, FileAccess.Read, FileShare.Read, data.Length, true);
//Task<int> returns the number of bytes read
Task<int> task = Task<int>.Factory.FromAsync(
fs.BeginRead, fs.EndRead, data, 0, data.Length, null);
// It is possible to do other work here while waiting
// for the antecedent task to complete.
// ...
// Add the continuation, which returns a Task<string>.
return task.ContinueWith((antecedent) =>
{
fs.Close();
// Result = "number of bytes read" (if we need it.)
if (antecedent.Result < 100)
{
return "Data is too small to bother with.";
}
else
{
// If we did not receive the entire file, the end of the
// data buffer will contain garbage.
if (antecedent.Result < data.Length)
Array.Resize(ref data, antecedent.Result);
// Will be returned in the Result property of the Task<string>
// at some future point after the asynchronous file I/O operation completes.
return new UTF8Encoding().GetString(data);
}
});
}
然後就可以呼叫這個方法，如下所示。
Dim myTask As Task(Of String) = GetFileStringAsync(path)
' Do some other work
' ...
Try
Console.WriteLine(myTask.Result.Substring(0, 500))
Catch ex As AggregateException
Console.WriteLine(ex.InnerException.Message)
End Try
Task<string> t = GetFileStringAsync(path);
// Do some other work:
// ...
try
{
Console.WriteLine(t.Result.Substring(0, 500));
}
catch (AggregateException ae)
{
Console.WriteLine(ae.InnerException.Message);
}
提供自訂狀態資料
在一般的 IAsyncResult 作業中，如果 AsyncCallback 委派需要一些自訂的狀態資料，您必須透過 Begin 方法的最後一個參數傳遞這些資訊，以便將資料封裝在成最後會傳遞給回呼方法的 IAsyncResult 物件中。 通常使用 FromAsync 方法時，並不需要這樣做。 如果接續工作已經知道這些自訂資料，就可以直接在接續工作委派中擷取這些資料。 下列範例與前一個範例類似，但是接續工作會檢查接續工作的使用者委派可直接存取的自訂狀態資料，而不是檢查前項工作的 Result 屬性。
Public Function GetFileStringAsync2(ByVal path As String) As Task(Of String)
Dim fi = New FileInfo(path)
Dim data(fi.Length) As Byte
Dim state As New MyCustomState()
Dim fs As New FileStream(path, FileMode.Open, FileAccess.Read, FileShare.Read, data.Length, True)
' We still pass null for the last parameter because
' the state variable is visible to the continuation delegate.
Dim myTask As Task(Of Integer) = Task(Of Integer).Factory.FromAsync(
AddressOf fs.BeginRead, AddressOf fs.EndRead, data, 0, data.Length, Nothing)
Return myTask.ContinueWith(Function(antecedent)
fs.Close()
' Capture custom state data directly in the user delegate.
' No need to pass it through the FromAsync method.
If (state.StateData.Contains("New York, New York")) Then
Return "Start spreading the news!"
End If
' If we did not receive the entire file, the end of the
' data buffer will contain garbage.
If (antecedent.Result < data.Length) Then
Array.Resize(data, antecedent.Result)
End If
'/ Will be returned in the Result property of the Task<string>
'/ at some future point after the asynchronous file I/O operation completes.
Return New UTF8Encoding().GetString(data)
End Function)
End Function
public Task<string> GetFileStringAsync2(string path)
{
FileInfo fi = new FileInfo(path);
byte[] data = new byte[fi.Length];
MyCustomState state = GetCustomState();
FileStream fs = new FileStream(path, FileMode.Open, FileAccess.Read, FileShare.Read, data.Length, true);
// We still pass null for the last parameter because
// the state variable is visible to the continuation delegate.
Task<int> task = Task<int>.Factory.FromAsync(
fs.BeginRead, fs.EndRead, data, 0, data.Length, null);
return task.ContinueWith((antecedent) =>
{
// It is safe to close the filestream now.
fs.Close();
// Capture custom state data directly in the user delegate.
// No need to pass it through the FromAsync method.
if (state.StateData.Contains("New York, New York"))
{
return "Start spreading the news!";
}
else
{
// If we did not receive the entire file, the end of the
// data buffer will contain garbage.
if (antecedent.Result < data.Length)
Array.Resize(ref data, antecedent.Result);
// Will be returned in the Result property of the Task<string>
// at some future point after the asynchronous file I/O operation completes.
return new UTF8Encoding().GetString(data);
}
});
}
同步處理多個 FromAsync 工作
靜態 ContinueWhenAll 和 ContinueWhenAny 方法在結合 FromAsync 方法一起使用時，可以提供額外的彈性。 下列範例顯示如何啟始多個非同步 I/O 作業，然後等它們全部完成之後，您再執行接續工作。
Public Function GetMultiFileData(ByVal filesToRead As String()) As Task(Of String)
Dim fs As FileStream
Dim tasks(filesToRead.Length) As Task(Of String)
Dim fileData() As Byte = Nothing
For i As Integer = 0 To filesToRead.Length
fileData(&H1000) = New Byte()
fs = New FileStream(filesToRead(i), FileMode.Open, FileAccess.Read, FileShare.Read, fileData.Length, True)
' By adding the continuation here, the
' Result of each task will be a string.
tasks(i) = Task(Of Integer).Factory.FromAsync(AddressOf fs.BeginRead,
AddressOf fs.EndRead,
fileData,
0,
fileData.Length,
Nothing).
ContinueWith(Function(antecedent)
fs.Close()
'If we did not receive the entire file, the end of the
' data buffer will contain garbage.
If (antecedent.Result < fileData.Length) Then
ReDim Preserve fileData(antecedent.Result)
End If
'Will be returned in the Result property of the Task<string>
' at some future point after the asynchronous file I/O operation completes.
Return New UTF8Encoding().GetString(fileData)
End Function)
Next
Return Task(Of String).Factory.ContinueWhenAll(tasks, Function(data)
' Propagate all exceptions and mark all faulted tasks as observed.
Task.WaitAll(data)
' Combine the results from all tasks.
Dim sb As New StringBuilder()
For Each t As Task(Of String) In data
sb.Append(t.Result)
Next
' Final result to be returned eventually on the calling thread.
Return sb.ToString()
End Function)
End Function
public Task<string> GetMultiFileData(string[] filesToRead)
{
FileStream fs;
Task<string>[] tasks = new Task<string>[filesToRead.Length];
byte[] fileData = null;
for (int i = 0; i < filesToRead.Length; i++)
{
fileData = new byte[0x1000];
fs = new FileStream(filesToRead[i], FileMode.Open, FileAccess.Read, FileShare.Read, fileData.Length, true);
// By adding the continuation here, the
// Result of each task will be a string.
tasks[i] = Task<int>.Factory.FromAsync(
fs.BeginRead, fs.EndRead, fileData, 0, fileData.Length, null)
.ContinueWith((antecedent) =>
{
fs.Close();
// If we did not receive the entire file, the end of the
// data buffer will contain garbage.
if (antecedent.Result < fileData.Length)
Array.Resize(ref fileData, antecedent.Result);
// Will be returned in the Result property of the Task<string>
// at some future point after the asynchronous file I/O operation completes.
return new UTF8Encoding().GetString(fileData);
});
}
// Wait for all tasks to complete.
return Task<string>.Factory.ContinueWhenAll(tasks, (data) =>
{
// Propagate all exceptions and mark all faulted tasks as observed.
Task.WaitAll(data);
// Combine the results from all tasks.
StringBuilder sb = new StringBuilder();
foreach (var t in data)
{
sb.Append(t.Result);
}
// Final result to be returned eventually on the calling thread.
return sb.ToString();
});
}
只用於 End 方法的 FromAsync 工作
在 Begin 方法需要多於三個參數或是具有 ref 或 out 參數的少數案例中，您可以使用只表示 End 方法的 FromAsync 多載，例如 TaskFactory<TResult>.FromAsync(IAsyncResult, Func<IAsyncResult, TResult>)。 在任何情況下，只要您接到 IAsyncResult 且想要將它封裝在 Task 中，也可以使用這些方法。
Shared Function ReturnTaskFromAsyncResult() As Task(Of String)
Dim ar As IAsyncResult = DoSomethingAsynchronously()
Dim t As Task(Of String) = Task(Of String).Factory.FromAsync(ar, Function(res) CStr(res.AsyncState))
Return t
End Function
static Task<String> ReturnTaskFromAsyncResult()
{
IAsyncResult ar = DoSomethingAsynchronously();
Task<String> t = Task<string>.Factory.FromAsync(ar, _ =>
{
return (string)ar.AsyncState;
});
return t;
}
啟動和取消 FromAsync 工作
FromAsync 方法傳回的工作會具有 WaitingForActivation 狀態，表示系統會在工作建立之後的某個時間點啟動工作。 如果您嘗試在這類工作上呼叫 Start，將會引發例外狀況。
您無法取消 FromAsync 工作，因為基礎的.NET Framework API 目前並不支援取消進行中的檔案或網路 I/O。 您可以在封裝 FromAsync 呼叫的方法中加入取消功能，但您只能在呼叫 FromAsync 之前或它完成之後 (例如在接續工作中) 對取消進行回應。
有些支援 EAP 的類別 (例如 WebClient) 可支援取消，您可以使用取消語彙基元來整合該原生取消功能。
將複雜的 EAP 作業公開為工作
不像 FromAsync 系列方法會包裝 IAsyncResult 模式，TPL 並不提供任何專為封裝事件架構非同步作業而設計的方法。 但是，TPL 提供 System.Threading.Tasks.TaskCompletionSource<TResult> 類別，這個類別可將任意一組作業以 Task<TResult> 表示。 這些作業可以是同步或非同步作業，也可以是 I/O 繫結作業、計算繫結作業或兩者都是。
下列範例顯示如何使用 TaskCompletionSource<TResult>，將一組非同步 WebClient 作業以基本 Task 的形式公開給用戶端程式碼。 這個方法可讓您輸入一組 Web URL，以及要搜尋的詞彙或名稱，然後傳回搜尋詞彙在每個網站上出現的次數。
Class SimpleWebExample
Dim tcs As New TaskCompletionSource(Of String())
Dim nameToSearch As String
Dim token As CancellationToken
Dim results As New List(Of String)
Dim m_lock As Object
Dim count As Integer
Dim addresses() As String
Public Function GetWordCountsSimplified(ByVal urls() As String, ByVal str As String, ByVal token As CancellationToken) As Task(Of String())
Dim webClients() As WebClient
ReDim webClients(urls.Length)
' If the user cancels the CancellationToken, then we can use the
' WebClient's ability to cancel its own async operations.
token.Register(Sub()
For Each wc As WebClient In webClients
If Not wc Is Nothing Then
wc.CancelAsync()
End If
Next
End Sub)
For i As Integer = 0 To urls.Length
webClients(i) = New WebClient()
' Specify the callback for the DownloadStringCompleted
' event that will be raised by this WebClient instance.
AddHandler webClients(i).DownloadStringCompleted, AddressOf WebEventHandler
Dim address As New Uri(urls(i))
' Pass the address, and also use it for the userToken
' to identify the page when the delegate is invoked.
webClients(i).DownloadStringAsync(address, address)
Next
' Return the underlying Task. The client code
' waits on the Result property, and handles exceptions
' in the try-catch block there.
Return tcs.Task
End Function
Public Sub WebEventHandler(ByVal sender As Object, ByVal args As DownloadStringCompletedEventArgs)
If args.Cancelled = True Then
tcs.TrySetCanceled()
Return
ElseIf Not args.Error Is Nothing Then
tcs.TrySetException(args.Error)
Return
Else
' Split the string into an array of words,
' then count the number of elements that match
' the search term.
Dim words() As String = args.Result.Split(" "c)
Dim NAME As String = nameToSearch.ToUpper()
Dim nameCount = (From word In words.AsParallel()
Where word.ToUpper().Contains(NAME)
Select word).Count()
' Associate the results with the url, and add new string to the array that
' the underlying Task object will return in its Result property.
results.Add(String.Format("{0} has {1} instances of {2}", args.UserState, nameCount, NAME))
End If
SyncLock (m_lock)
count = count + 1
If (count = addresses.Length) Then
tcs.TrySetResult(results.ToArray())
End If
End SyncLock
End Sub
End Class
Task<string[]> GetWordCountsSimplified(string[] urls, string name, CancellationToken token)
{
TaskCompletionSource<string[]> tcs = new TaskCompletionSource<string[]>();
WebClient[] webClients = new WebClient[urls.Length];
object m_lock = new object();
int count = 0;
List<string> results = new List<string>();
// If the user cancels the CancellationToken, then we can use the
// WebClient's ability to cancel its own async operations.
token.Register(() =>
{
foreach (var wc in webClients)
{
if (wc != null)
wc.CancelAsync();
}
});
for (int i = 0; i < urls.Length; i++)
{
webClients[i] = new WebClient();
#region callback
// Specify the callback for the DownloadStringCompleted
// event that will be raised by this WebClient instance.
webClients[i].DownloadStringCompleted += (obj, args) =>
{
// Argument validation and exception handling omitted for brevity.
// Split the string into an array of words,
// then count the number of elements that match
// the search term.
string[] words = args.Result.Split(' ');
string NAME = name.ToUpper();
int nameCount = (from word in words.AsParallel()
where word.ToUpper().Contains(NAME)
select word)
.Count();
// Associate the results with the url, and add new string to the array that
// the underlying Task object will return in its Result property.
results.Add(String.Format("{0} has {1} instances of {2}", args.UserState, nameCount, name));
// If this is the last async operation to complete,
// then set the Result property on the underlying Task.
lock (m_lock)
{
count++;
if (count == urls.Length)
{
tcs.TrySetResult(results.ToArray());
}
}
};
#endregion
// Call DownloadStringAsync for each URL.
Uri address = null;
address = new Uri(urls[i]);
webClients[i].DownloadStringAsync(address, address);
} // end for
// Return the underlying Task. The client code
// waits on the Result property, and handles exceptions
// in the try-catch block there.
return tcs.Task;
}
如需更完整的範例，包括其他例外處理以及示範如何從用戶端程式碼呼叫方法，請參閱 HOW TO：將 EAP 模式包裝在工作中。
請記住，TaskCompletionSource<TResult> 建立的所有工作都會由該 TaskCompletionSource 啟動，因此，使用者程式碼不應該在該工作上呼叫 Start 方法。
使用工作來實作 APM 模式
在某些情節中，可能需要使用 API 中的 Begin/End 方法配對來直接公開 IAsyncResult 模式。 例如，您可能需要維護與現有 API 之間的一致性，或是您可能有需要這種模式的自動化工具。 在這種情況下，您可以使用 Task 來簡化內部實作 APM 模式的方式。
下列範例顯示如何針對長時間執行的計算繫結方法，使用工作來實作 APM Begin/End 方法配對。
Class Calculator
Public Function BeginCalculate(ByVal decimalPlaces As Integer, ByVal ac As AsyncCallback, ByVal state As Object) As IAsyncResult
Console.WriteLine("Calling BeginCalculate on thread {0}", Thread.CurrentThread.ManagedThreadId)
Dim myTask = Task(Of String).Factory.StartNew(Function(obj) Compute(decimalPlaces), state)
myTask.ContinueWith(Sub(antedecent) ac(myTask))
End Function
Private Function Compute(ByVal decimalPlaces As Integer)
Console.WriteLine("Calling compute on thread {0}", Thread.CurrentThread.ManagedThreadId)
' Simulating some heavy work.
Thread.SpinWait(500000000)
' Actual implemenation left as exercise for the reader.
' Several examples are available on the Web.
Return "3.14159265358979323846264338327950288"
End Function
Public Function EndCalculate(ByVal ar As IAsyncResult) As String
Console.WriteLine("Calling EndCalculate on thread {0}", Thread.CurrentThread.ManagedThreadId)
Return CType(ar, Task(Of String)).Result
End Function
End Class
Class CalculatorClient
Shared decimalPlaces As Integer
Shared Sub Main()
Dim calc As New Calculator
Dim places As Integer = 35
Dim callback As New AsyncCallback(AddressOf PrintResult)
Dim ar As IAsyncResult = calc.BeginCalculate(places, callback, calc)
' Do some work on this thread while the calulator is busy.
Console.WriteLine("Working...")
Thread.SpinWait(500000)
Console.ReadLine()
End Sub
Public Shared Sub PrintResult(ByVal result As IAsyncResult)
Dim c As Calculator = CType(result.AsyncState, Calculator)
Dim piString As String = c.EndCalculate(result)
Console.WriteLine("Calling PrintResult on thread {0}; result = {1}",
Thread.CurrentThread.ManagedThreadId, piString)
End Sub
End Class
class Calculator
{
public IAsyncResult BeginCalculate(int decimalPlaces, AsyncCallback ac, object state)
{
Console.WriteLine("Calling BeginCalculate on thread {0}", Thread.CurrentThread.ManagedThreadId);
Task<string> f = Task<string>.Factory.StartNew(_ => Compute(decimalPlaces), state);
if (ac != null) f.ContinueWith((res) => ac(f));
return f;
}
public string Compute(int numPlaces)
{
Console.WriteLine("Calling compute on thread {0}", Thread.CurrentThread.ManagedThreadId);
// Simulating some heavy work.
Thread.SpinWait(500000000);
// Actual implemenation left as exercise for the reader.
// Several examples are available on the Web.
return "3.14159265358979323846264338327950288";
}
public string EndCalculate(IAsyncResult ar)
{
Console.WriteLine("Calling EndCalculate on thread {0}", Thread.CurrentThread.ManagedThreadId);
return ((Task<string>)ar).Result;
}
}
public class CalculatorClient
{
static int decimalPlaces = 12;
public static void Main()
{
Calculator calc = new Calculator();
int places = 35;
AsyncCallback callBack = new AsyncCallback(PrintResult);
IAsyncResult ar = calc.BeginCalculate(places, callBack, calc);
// Do some work on this thread while the calulator is busy.
Console.WriteLine("Working...");
Thread.SpinWait(500000);
Console.ReadLine();
}
public static void PrintResult(IAsyncResult result)
{
Calculator c = (Calculator)result.AsyncState;
string piString = c.EndCalculate(result);
Console.WriteLine("Calling PrintResult on thread {0}; result = {1}",
Thread.CurrentThread.ManagedThreadId, piString);
}
}
使用 StreamExtensions 範例程式碼
MSDN 網站上使用 .NET Framework 4 進行平行程式設計的範例 (英文) 中的 Streamextensions.cs 檔案包含數個參考實作，這些參考實作會使用 Task 物件來執行非同步檔案和網路 I/O。