Como: Executar ação quando um bloco de fluxo de dados recebe dados
Os tipos de bloco de fluxo de dados de execução chamam um delegado fornecido pelo usuário quando eles recebem dados. As System.Threading.Tasks.Dataflow.ActionBlock<TInput>classes , System.Threading.Tasks.Dataflow.TransformBlock<TInput,TOutput>e são System.Threading.Tasks.Dataflow.TransformManyBlock<TInput,TOutput> tipos de bloco de fluxo de dados de execução. Você pode usar a delegate palavra-chave (Sub no Visual Basic), Action<T>, Func<T,TResult>, ou uma expressão lambda quando você fornece uma função de trabalho para um bloco de fluxo de dados de execução. Este documento descreve como usar Func<T,TResult> expressões lambda e lambda para executar ações em blocos de execução.
Nota
A biblioteca de fluxo de dados TPL (o namespace) não é distribuída com o System.Threading.Tasks.Dataflow .NET. Para instalar o System.Threading.Tasks.Dataflow namespace no Visual Studio, abra seu projeto, escolha Gerenciar pacotes NuGet no menu Projeto e pesquise o System.Threading.Tasks.Dataflow pacote online. Como alternativa, para instalá-lo usando a CLI do .NET Core, execute dotnet add package System.Threading.Tasks.Dataflow.
Exemplo
O exemplo a seguir usa o fluxo de dados para ler um arquivo do disco e calcula o número de bytes nesse arquivo que são iguais a zero. Ele usa TransformBlock<TInput,TOutput> para ler o arquivo e calcular o número de zero bytes, e ActionBlock<TInput> para imprimir o número de zero bytes para o console. O TransformBlock<TInput,TOutput> objeto especifica um Func<T,TResult> objeto para executar o trabalho quando os blocos recebem dados. O ActionBlock<TInput> objeto usa uma expressão lambda para imprimir no console o número de zero bytes que são lidos.
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.
'
Embora você possa fornecer uma expressão lambda para um TransformBlock<TInput,TOutput> objeto, este exemplo usa Func<T,TResult> para permitir que outro código use o CountBytes método. O ActionBlock<TInput> objeto usa uma expressão lambda porque o trabalho a ser executado é específico para essa tarefa e provavelmente não será útil de outro código. Para obter mais informações sobre como as expressões lambda funcionam na Biblioteca Paralela de Tarefas, consulte Expressões do Lambda em PLINQ e TPL.
A seção Resumo dos Tipos de Delegados no documento Fluxo de Dados resume os tipos de delegados que você pode fornecer a ActionBlock<TInput>, TransformBlock<TInput,TOutput>e TransformManyBlock<TInput,TOutput> objetos. A tabela também especifica se o tipo de delegado opera de forma síncrona ou assíncrona.
Programação robusta
Este exemplo fornece um delegado de tipo Func<T,TResult> ao TransformBlock<TInput,TOutput> objeto para executar a tarefa do bloco de fluxo de dados de forma síncrona. Para permitir que o bloco de fluxo de dados se comporte de forma assíncrona, forneça um delegado do tipo Func<T, Task<TResult>> para o bloco de fluxo de dados. Quando um bloco de fluxo de dados se comporta de forma assíncrona, a tarefa do bloco de fluxo de dados é concluída somente quando o objeto retornado Task<TResult> é concluído. O exemplo a seguir modifica o CountBytes método e usa os operadores async e await (Async e Await no Visual Basic) para calcular de forma assíncrona o número total de bytes que são zero no arquivo fornecido. O ReadAsync método executa operações de leitura de arquivo de forma assíncrona.
// 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
Você também pode usar expressões lambda assíncronas para executar ações em um bloco de fluxo de dados de execução. O exemplo a seguir modifica o TransformBlock<TInput,TOutput> objeto usado no exemplo anterior para que ele use uma expressão lambda para executar o trabalho de forma assíncrona.
// 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)
