Postupy: Provádění akcí po přijetí dat do bloku toku dat
Typy bloku toku dat spouštění volají delegáta poskytnutého uživatelem při příjmu dat. Třídy System.Threading.Tasks.Dataflow.ActionBlock<TInput>a třídy System.Threading.Tasks.Dataflow.TransformBlock<TInput,TOutput>System.Threading.Tasks.Dataflow.TransformManyBlock<TInput,TOutput> jsou typy bloku toku dat spouštění. Klíčové slovo (Subv jazyce Visual Basic), Action<T>, Func<T,TResult>nebo výraz lambda můžete použít delegate při poskytování pracovní funkce bloku toku dat provádění. Tento dokument popisuje, jak používat Func<T,TResult> výrazy lambda k provádění akcí v blocích provádění.
Poznámka:
Knihovna toku dat TPL ( System.Threading.Tasks.Dataflow obor názvů) není distribuována s .NET. Pokud chcete nainstalovat System.Threading.Tasks.Dataflow obor názvů v sadě Visual Studio, otevřete projekt, v nabídce Projekt zvolte Spravovat balíčky NuGet a vyhledejte System.Threading.Tasks.Dataflow balíček online. Pokud ho chcete nainstalovat pomocí rozhraní příkazového řádku .NET Core, spusťte dotnet add package System.Threading.Tasks.Dataflow.
Příklad
Následující příklad používá tok dat ke čtení souboru z disku a vypočítá počet bajtů v tomto souboru, které jsou rovny nule. Používá TransformBlock<TInput,TOutput> se ke čtení souboru a výpočtu počtu bajtů nula a ActionBlock<TInput> k vytištění počtu bajtů v konzole. Objekt TransformBlock<TInput,TOutput> určuje objekt, který Func<T,TResult> má provádět práci, když bloky přijímají data. Objekt ActionBlock<TInput> používá výraz lambda k tisku do konzoly počet nul bajtů, které jsou přečteny.
using System;
using System.IO;
using System.Linq;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;
// Demonstrates how to provide delegates to exectution dataflow blocks.
class DataflowExecutionBlocks
{
// Computes the number of zero bytes that the provided file
// contains.
static int CountBytes(string path)
{
byte[] buffer = new byte[1024];
int totalZeroBytesRead = 0;
using (var fileStream = File.OpenRead(path))
{
int bytesRead = 0;
do
{
bytesRead = fileStream.Read(buffer, 0, buffer.Length);
totalZeroBytesRead += buffer.Count(b => b == 0);
} while (bytesRead > 0);
}
return totalZeroBytesRead;
}
static void Main(string[] args)
{
// Create a temporary file on disk.
string tempFile = Path.GetTempFileName();
// Write random data to the temporary file.
using (var fileStream = File.OpenWrite(tempFile))
{
Random rand = new Random();
byte[] buffer = new byte[1024];
for (int i = 0; i < 512; i++)
{
rand.NextBytes(buffer);
fileStream.Write(buffer, 0, buffer.Length);
}
}
// Create an ActionBlock<int> object that prints to the console
// the number of bytes read.
var printResult = new ActionBlock<int>(zeroBytesRead =>
{
Console.WriteLine("{0} contains {1} zero bytes.",
Path.GetFileName(tempFile), zeroBytesRead);
});
// Create a TransformBlock<string, int> object that calls the
// CountBytes function and returns its result.
var countBytes = new TransformBlock<string, int>(
new Func<string, int>(CountBytes));
// Link the TransformBlock<string, int> object to the
// ActionBlock<int> object.
countBytes.LinkTo(printResult);
// Create a continuation task that completes the ActionBlock<int>
// object when the TransformBlock<string, int> finishes.
countBytes.Completion.ContinueWith(delegate { printResult.Complete(); });
// Post the path to the temporary file to the
// TransformBlock<string, int> object.
countBytes.Post(tempFile);
// Requests completion of the TransformBlock<string, int> object.
countBytes.Complete();
// Wait for the ActionBlock<int> object to print the message.
printResult.Completion.Wait();
// Delete the temporary file.
File.Delete(tempFile);
}
}
/* Sample output:
tmp4FBE.tmp contains 2081 zero bytes.
*/
Imports System.IO
Imports System.Linq
Imports System.Threading.Tasks
Imports System.Threading.Tasks.Dataflow
' Demonstrates how to provide delegates to exectution dataflow blocks.
Friend Class DataflowExecutionBlocks
' Computes the number of zero bytes that the provided file
' contains.
Private Shared Function CountBytes(ByVal path As String) As Integer
Dim buffer(1023) As Byte
Dim totalZeroBytesRead As Integer = 0
Using fileStream = File.OpenRead(path)
Dim bytesRead As Integer = 0
Do
bytesRead = fileStream.Read(buffer, 0, buffer.Length)
totalZeroBytesRead += buffer.Count(Function(b) b = 0)
Loop While bytesRead > 0
End Using
Return totalZeroBytesRead
End Function
Shared Sub Main(ByVal args() As String)
' Create a temporary file on disk.
Dim tempFile As String = Path.GetTempFileName()
' Write random data to the temporary file.
Using fileStream = File.OpenWrite(tempFile)
Dim rand As New Random()
Dim buffer(1023) As Byte
For i As Integer = 0 To 511
rand.NextBytes(buffer)
fileStream.Write(buffer, 0, buffer.Length)
Next i
End Using
' Create an ActionBlock<int> object that prints to the console
' the number of bytes read.
Dim printResult = New ActionBlock(Of Integer)(Sub(zeroBytesRead) Console.WriteLine("{0} contains {1} zero bytes.", Path.GetFileName(tempFile), zeroBytesRead))
' Create a TransformBlock<string, int> object that calls the
' CountBytes function and returns its result.
Dim countBytes = New TransformBlock(Of String, Integer)(New Func(Of String, Integer)(AddressOf DataflowExecutionBlocks.CountBytes))
' Link the TransformBlock<string, int> object to the
' ActionBlock<int> object.
countBytes.LinkTo(printResult)
' Create a continuation task that completes the ActionBlock<int>
' object when the TransformBlock<string, int> finishes.
countBytes.Completion.ContinueWith(Sub() printResult.Complete())
' Post the path to the temporary file to the
' TransformBlock<string, int> object.
countBytes.Post(tempFile)
' Requests completion of the TransformBlock<string, int> object.
countBytes.Complete()
' Wait for the ActionBlock<int> object to print the message.
printResult.Completion.Wait()
' Delete the temporary file.
File.Delete(tempFile)
End Sub
End Class
' Sample output:
'tmp4FBE.tmp contains 2081 zero bytes.
'
I když můžete objektu TransformBlock<TInput,TOutput> poskytnout výraz lambda, tento příklad používá Func<T,TResult> k povolení použití metody jiným kódem CountBytes . Objekt ActionBlock<TInput> používá výraz lambda, protože práce, která se má provést, je specifická pro tuto úlohu a pravděpodobně nebude užitečná z jiného kódu. Další informace o tom, jak výrazy lambda fungují v paralelní knihovně úloh, naleznete v tématu Výrazy lambda v PLINQ a TPL.
Část Souhrn typů delegátů v dokumentu Tok dat shrnuje typy delegátů, které můžete zadat pro ActionBlock<TInput>, TransformBlock<TInput,TOutput>a TransformManyBlock<TInput,TOutput> objekty. Tabulka také určuje, jestli typ delegáta funguje synchronně nebo asynchronně.
Robustní programování
Tento příklad poskytuje delegáta typu Func<T,TResult> objektu TransformBlock<TInput,TOutput> k provedení úlohy bloku toku dat synchronně. Pokud chcete povolit, aby se blok toku dat choval asynchronně, zadejte delegáta typu Func<T, Task<TResult>> do bloku toku dat. Když se blok toku dat chová asynchronně, je úloha bloku toku dat dokončena pouze v případě, že se vrácený Task<TResult> objekt dokončí. Následující příklad upraví metodu CountBytes a použije asynchronní operátory a operátory await (Async a Await v jazyce Visual Basic) k asynchronnímu výpočtu celkového počtu bajtů, které jsou v zadaném souboru nuly. Metoda ReadAsync provádí operace čtení souborů asynchronně.
// Asynchronously computes the number of zero bytes that the provided file
// contains.
static async Task<int> CountBytesAsync(string path)
{
byte[] buffer = new byte[1024];
int totalZeroBytesRead = 0;
using (var fileStream = new FileStream(
path, FileMode.Open, FileAccess.Read, FileShare.Read, 0x1000, true))
{
int bytesRead = 0;
do
{
// Asynchronously read from the file stream.
bytesRead = await fileStream.ReadAsync(buffer, 0, buffer.Length);
totalZeroBytesRead += buffer.Count(b => b == 0);
} while (bytesRead > 0);
}
return totalZeroBytesRead;
}
' Asynchronously computes the number of zero bytes that the provided file
' contains.
Private Shared async Function CountBytesAsync(ByVal path As String) As Task(Of Integer)
Dim buffer(1023) As Byte
Dim totalZeroBytesRead As Integer = 0
Using fileStream = New FileStream(path, FileMode.Open, FileAccess.Read, FileShare.Read, &H1000, True)
Dim bytesRead As Integer = 0
Do
' Asynchronously read from the file stream.
bytesRead = await fileStream.ReadAsync(buffer, 0, buffer.Length)
totalZeroBytesRead += buffer.Count(Function(b) b = 0)
Loop While bytesRead > 0
End Using
Return totalZeroBytesRead
End Function
Asynchronní výrazy lambda můžete také použít k provedení akce v bloku toku dat provádění. Následující příklad upraví TransformBlock<TInput,TOutput> objekt, který se používá v předchozím příkladu, aby používal výraz lambda k provedení práce asynchronně.
// Create a TransformBlock<string, int> object that calls the
// CountBytes function and returns its result.
var countBytesAsync = new TransformBlock<string, int>(async path =>
{
byte[] buffer = new byte[1024];
int totalZeroBytesRead = 0;
using (var fileStream = new FileStream(
path, FileMode.Open, FileAccess.Read, FileShare.Read, 0x1000, true))
{
int bytesRead = 0;
do
{
// Asynchronously read from the file stream.
bytesRead = await fileStream.ReadAsync(buffer, 0, buffer.Length);
totalZeroBytesRead += buffer.Count(b => b == 0);
} while (bytesRead > 0);
}
return totalZeroBytesRead;
});
' Create a TransformBlock<string, int> object that calls the
' CountBytes function and returns its result.
Dim countBytesAsync = New TransformBlock(Of String, Integer)(async Function(path)
' Asynchronously read from the file stream.
Dim buffer(1023) As Byte
Dim totalZeroBytesRead As Integer = 0
Using fileStream = New FileStream(path, FileMode.Open, FileAccess.Read, FileShare.Read, &H1000, True)
Dim bytesRead As Integer = 0
Do
bytesRead = await fileStream.ReadAsync(buffer, 0, buffer.Length)
totalZeroBytesRead += buffer.Count(Function(b) b = 0)
Loop While bytesRead > 0
End Using
Return totalZeroBytesRead
End Function)
