TPL Dataflow | CSharp

TPL Dataflow | CSharp

O TPL Dataflow (Task Parallel Library Dataflow) é uma biblioteca da .NET — parte do namespace System.Threading.Tasks.Dataflow — projetada para criar pipelines e sistemas concorrentes baseados em mensagens (dataflow programming).

🔹 O que é o TPL Dataflow?

O TPL Dataflow é uma abstração de alto nível sobre tasks e concorrência, criada para simplificar o processamento assíncrono e paralelo de dados. Em vez de você gerenciar threads, locks ou filas manualmente, o TPL Dataflow oferece blocos de processamento conectáveis, que passam dados entre si de forma segura e eficiente.

Ele é distribuído como um pacote NuGet:

dotnet add package System.Threading.Tasks.Dataflow

⚙️ Conceito central: Blocos (Blocks)

Um dataflow block é uma unidade de processamento. Existem três categorias principais:

1. Fonte (Source Blocks)

Produzem dados. ➡️ Exemplo: BufferBlock<T> — armazena mensagens até serem processadas.

2. Alvo (Target Blocks)

Recebem e consomem dados. ➡️ Exemplo: ActionBlock<T> — executa uma ação para cada item recebido.

3. Transformadores (Transform Blocks)

Recebem dados, processam e enviam o resultado a outro bloco. ➡️ Exemplo: TransformBlock<TInput, TOutput> — transforma os dados de entrada em saída.

🔄 Exemplo prático

Imagine um pipeline que:

Recebe números.
Multiplica cada um por 2.
Exibe o resultado.

using System;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;

class Program
{
    static async Task Main()
    {
        // 1️⃣ Cria um bloco de transformação (multiplica por 2)
        var multiplyBlock = new TransformBlock<int, int>(n =>
        {
            Console.WriteLine($"Multiplicando {n}...");
            return n * 2;
        });

        // 2️⃣ Cria um bloco de ação (imprime resultado)
        var printBlock = new ActionBlock<int>(n =>
        {
            Console.WriteLine($"Resultado: {n}");
        });

        // 3️⃣ Conecta os blocos
        multiplyBlock.LinkTo(printBlock, new DataflowLinkOptions { PropagateCompletion = true });

        // 4️⃣ Posta dados
        for (int i = 1; i <= 5; i++)
            multiplyBlock.Post(i);

        // 5️⃣ Finaliza o pipeline
        multiplyBlock.Complete();
        await printBlock.Completion;
    }
}

🧠 O que acontece:

Cada número é enviado ao multiplyBlock.
O resultado é passado automaticamente ao printBlock.
Tudo isso de forma assíncrona e segura para múltiplas threads.

⚡ Recursos avançados

Controle de concorrência: Limite de tarefas simultâneas (MaxDegreeOfParallelism).
Propagação de falhas e cancelamentos: PropagateCompletion = true e CancellationToken.
Bufferização e backpressure: O TPL Dataflow gerencia o acúmulo de mensagens automaticamente.
Composição de pipelines complexos: Você pode conectar vários blocos (transformações, filtros, junções, etc.) para processar fluxos de dados grandes.

🧩 Exemplo com paralelismo e cancelamento

var options = new ExecutionDataflowBlockOptions
{
    MaxDegreeOfParallelism = 4,
    CancellationToken = token
};

var block = new ActionBlock<int>(async n =>
{
    await Task.Delay(1000);
    Console.WriteLine($"Processando {n}");
}, options);

Isso cria um bloco que processa até 4 itens simultaneamente, respeitando um token de cancelamento.

🏁 Quando usar TPL Dataflow

✅ Ideal para:

Pipelines de processamento (ETL, stream, filas internas).
Processamento paralelo de mensagens/eventos.
Substituir filas como BlockingCollection com mais flexibilidade.

❌ Não ideal para:

Casos simples onde async/await ou Parallel.ForEach já resolvem.
Comunicação entre processos (use filas como RabbitMQ ou Kafka para isso).

Exemplo de Uso

Imagine que você tem uma pasta com vários arquivos grandes. Queremos processá-los em três etapas:

Ler cada arquivo do disco.
Comprimir (simulado).
Enviar para um servidor (simulado).

E tudo isso de forma paralela, mas controlando quantas tarefas rodam ao mesmo tempo.

⚙️ Código completo

using System;
using System.IO;
using System.IO.Compression;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;

class Program
{
    static async Task Main()
    {
        var files = Directory.GetFiles(@"C:\ArquivosParaProcessar");

        // 1️⃣ Bloco de leitura — lê o conteúdo do arquivo
        var readBlock = new TransformBlock<string, byte[]>(async path =>
        {
            Console.WriteLine($"📥 Lendo: {Path.GetFileName(path)}");
            return await File.ReadAllBytesAsync(path);
        },
        new ExecutionDataflowBlockOptions
        {
            MaxDegreeOfParallelism = 2 // lê até 2 arquivos simultaneamente
        });

        // 2️⃣ Bloco de compressão — comprime os bytes
        var compressBlock = new TransformBlock<byte[], byte[]>(async data =>
        {
            await Task.Delay(500); // simula tempo de compressão
            using var output = new MemoryStream();
            using (var gzip = new GZipStream(output, CompressionMode.Compress, true))
                gzip.Write(data, 0, data.Length);

            Console.WriteLine($"🗜️  Comprimido {data.Length / 1024} KB → {output.Length / 1024} KB");
            return output.ToArray();
        },
        new ExecutionDataflowBlockOptions
        {
            MaxDegreeOfParallelism = 4 // comprime até 4 arquivos simultaneamente
        });

        // 3️⃣ Bloco de upload — envia os dados simulando uma operação de rede
        var uploadBlock = new ActionBlock<byte[]>(async compressed =>
        {
            await Task.Delay(800); // simula upload
            Console.WriteLine($"☁️  Upload concluído ({compressed.Length / 1024} KB)");
        },
        new ExecutionDataflowBlockOptions
        {
            MaxDegreeOfParallelism = 3 // 3 uploads paralelos
        });

        // 4️⃣ Conecta os blocos
        readBlock.LinkTo(compressBlock, new DataflowLinkOptions { PropagateCompletion = true });
        compressBlock.LinkTo(uploadBlock, new DataflowLinkOptions { PropagateCompletion = true });

        // 5️⃣ Envia os arquivos para o pipeline
        foreach (var file in files)
            await readBlock.SendAsync(file);

        // 6️⃣ Sinaliza o fim do envio
        readBlock.Complete();

        // 7️⃣ Aguarda todo o pipeline finalizar
        await uploadBlock.Completion;

        Console.WriteLine("\n✅ Pipeline concluído!");
    }
}

🧠 O que está acontecendo

Etapa	Bloco	Responsabilidade	Paralelismo
1️⃣	`TransformBlock<string, byte[]>`	Lê arquivos do disco	2
2️⃣	`TransformBlock<byte[], byte[]>`	Comprime dados	4
3️⃣	`ActionBlock<byte[]>`	Faz upload	3

Cada bloco:

Recebe mensagens de entrada,
Processa de forma assíncrona,
Envia a saída para o próximo bloco,
E o Dataflow gerencia automaticamente buffers, backpressure e sincronização.

🧩 Vantagens

✅ Código limpo e modular — cada etapa é independente. ✅ Controle fino de concorrência (sem lock, Semaphore, etc.). ✅ Fácil composição de pipelines complexos. ✅ Cancelamento, retries e logs são fáceis de adicionar.

🛠️ Dica prática

Para usar:

dotnet add package System.Threading.Tasks.Dataflow

E no início do arquivo:

using System.Threading.Tasks.Dataflow;

🔹 O que é o TPL Dataflow?​

⚙️ Conceito central: Blocos (Blocks)​

1. Fonte (Source Blocks)​

2. Alvo (Target Blocks)​

3. Transformadores (Transform Blocks)​

🔄 Exemplo prático​

⚡ Recursos avançados​

🧩 Exemplo com paralelismo e cancelamento​

🏁 Quando usar TPL Dataflow​

Exemplo de Uso​

⚙️ Código completo​

🧠 O que está acontecendo​

🧩 Vantagens​

🛠️ Dica prática​