Como implementar um padrão de fluxo de dados produtor-consumidor
Neste artigo, você aprenderá como usar a biblioteca de fluxo de dados TPL para implementar um padrão produtor-consumidor. Nesse padrão, o produtor envia mensagens para um bloco de mensagens e o consumidor lê mensagens desse bloco.
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 demonstra um modelo básico produtor-consumidor que usa fluxo de dados. O Produce método grava matrizes que contêm bytes aleatórios de dados em um System.Threading.Tasks.Dataflow.ITargetBlock<TInput> objeto e o Consume método lê bytes de um System.Threading.Tasks.Dataflow.ISourceBlock<TOutput> objeto. Ao agir sobre as ISourceBlock<TOutput> interfaces e ITargetBlock<TInput> , em vez de seus tipos derivados, você pode escrever código reutilizável que pode atuar em uma variedade de tipos de bloco de fluxo de dados. Este exemplo usa a BufferBlock<T> classe. Como a BufferBlock<T> classe atua como um bloco de origem e como um bloco de destino, o produtor e o consumidor podem usar um objeto compartilhado para transferir dados.
O Produce método chama o Post método em um loop para gravar dados de forma síncrona no bloco de destino. Depois que o Produce método grava todos os dados no bloco de destino, ele chama o Complete método para indicar que o bloco nunca terá dados adicionais disponíveis. O Consume método usa os operadores async e await (Async e Await no Visual Basic) para calcular de forma assíncrona o número total de bytes recebidos do ISourceBlock<TOutput> objeto. Para agir de forma assíncrona, o Consume método chama o OutputAvailableAsync método para receber uma notificação quando o bloco de origem tem dados disponíveis e quando o bloco de origem nunca terá dados adicionais disponíveis.
using System;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;
class DataflowProducerConsumer
{
static void Produce(ITargetBlock<byte[]> target)
{
var rand = new Random();
for (int i = 0; i < 100; ++ i)
{
var buffer = new byte[1024];
rand.NextBytes(buffer);
target.Post(buffer);
}
target.Complete();
}
static async Task<int> ConsumeAsync(ISourceBlock<byte[]> source)
{
int bytesProcessed = 0;
while (await source.OutputAvailableAsync())
{
byte[] data = await source.ReceiveAsync();
bytesProcessed += data.Length;
}
return bytesProcessed;
}
static async Task Main()
{
var buffer = new BufferBlock<byte[]>();
var consumerTask = ConsumeAsync(buffer);
Produce(buffer);
var bytesProcessed = await consumerTask;
Console.WriteLine($"Processed {bytesProcessed:#,#} bytes.");
}
}
// Sample output:
// Processed 102,400 bytes.
Imports System.Threading.Tasks
Imports System.Threading.Tasks.Dataflow
Friend Class DataflowProducerConsumer
Private Shared Sub Produce(ByVal target As ITargetBlock(Of Byte()))
Dim rand As New Random()
For i As Integer = 0 To 99
Dim buffer(1023) As Byte
rand.NextBytes(buffer)
target.Post(buffer)
Next i
target.Complete()
End Sub
Private Shared Async Function ConsumeAsync(
ByVal source As ISourceBlock(Of Byte())) As Task(Of Integer)
Dim bytesProcessed As Integer = 0
Do While Await source.OutputAvailableAsync()
Dim data() As Byte = Await source.ReceiveAsync()
bytesProcessed += data.Length
Loop
Return bytesProcessed
End Function
Shared Sub Main()
Dim buffer = New BufferBlock(Of Byte())()
Dim consumer = ConsumeAsync(buffer)
Produce(buffer)
Dim result = consumer.GetAwaiter().GetResult()
Console.WriteLine($"Processed {result:#,#} bytes.")
End Sub
End Class
' Sample output:
' Processed 102,400 bytes.
Programação robusta
O exemplo anterior usa apenas um consumidor para processar os dados de origem. Se você tiver vários consumidores em seu aplicativo, use o TryReceive método para ler dados do bloco de origem, conforme mostrado no exemplo a seguir.
static async Task<int> ConsumeAsync(IReceivableSourceBlock<byte[]> source)
{
int bytesProcessed = 0;
while (await source.OutputAvailableAsync())
{
while (source.TryReceive(out byte[] data))
{
bytesProcessed += data.Length;
}
}
return bytesProcessed;
}
Private Shared Async Function ConsumeAsync(
ByVal source As IReceivableSourceBlock(Of Byte())) As Task(Of Integer)
Dim bytesProcessed As Integer = 0
Do While Await source.OutputAvailableAsync()
Dim data() As Byte
Do While source.TryReceive(data)
bytesProcessed += data.Length
Loop
Loop
Return bytesProcessed
End Function
O TryReceive método retorna False quando nenhum dado está disponível. Quando vários consumidores precisam acessar o bloco de origem simultaneamente, esse mecanismo garante que os dados ainda estejam disponíveis após a chamada para OutputAvailableAsync.
