Aufgabenwarteschlangen in C# (.NET): System.Threading.Channels und TPL Dataflow

Aufgabenwarteschlangen für C# (.NET): System.Threading.Channels und TPL Dataflow

Bei der Entwicklung von Multithread-Anwendungen in C# (.NET) besteht häufig die Notwendigkeit, eine asynchrone Datenübertragung zwischen Produzenten (producers) und Konsumenten (consumers) zu organisieren. Das klassische Producer-Consumer-Muster wird mit Hilfe von Aufgabenwarteschlangen gelöst. Im .NET-Ökosystem gibt es zwei leistungsstarke Werkzeuge dafür: System.Threading.Channels (eingeführt in .NET Core 3.0) und die Bibliothek TPL Dataflow (Teil von .NET seit 4.5).

In diesem Artikel werden wir beide Ansätze, ihre Besonderheiten und praktische Codebeispiele erläutern.

1. System.Threading.Channels — leichte und schnelle Warteschlange

Der Namespace System.Threading.Channels bietet threadsichere, asynchrone Warteschlangen, die für Szenarien mit hohem Durchsatz optimiert sind. Die wichtigsten Typen:

• Channel<T> — abstrakte Klasse eines Kanals.
• Channel.CreateUnbounded<T>() — unbegrenzte Warteschlange (Risiko von Speicherüberlauf).
• Channel.CreateBounded<T>(int capacity) — begrenzte Warteschlange mit Druckkontrolle.
• ChannelWriter<T> — Schnittstelle zum Schreiben.
• ChannelReader<T> — Schnittstelle zum Lesen.

Beispiel: Producer-Consumer mit Channel

Erstellen wir ein einfaches Beispiel, bei dem ein Produzent Zahlen generiert und ein Konsument diese verarbeitet.

using System.Threading.Channels;using System.Threading.Tasks;

var channel = Channel.CreateUnbounded<int>();

// Producervar producer = Task.Run(async () =>{    for (int i = 0; i < 10; i++)    {        await channel.Writer.WriteAsync(i);        Console.WriteLine($"Produziert: {i}");        await Task.Delay(100);    }    channel.Writer.Complete(); // Signal für das Ende});

// Consumervar consumer = Task.Run(async () =>{    await foreach (var item in channel.Reader.ReadAllAsync())    {        Console.WriteLine($"Verarbeitet: {item * 2}");        await Task.Delay(200);    }});

await Task.WhenAll(producer, consumer);Console.WriteLine("Fertig!");

Vorteile von Channels:

• Minimaler Overhead (keine Zuweisung zusätzlicher Objekte).
• Integrierte Unterstützung für async/await und IAsyncEnumerable.
• Ideal für hochbelastete Systeme (z. B. Verarbeitung von Netzwerkpaketen).

2. TPL Dataflow — leistungsstarke Verarbeitungspipeline

Die Bibliothek System.Threading.Tasks.Dataflow bietet komplexere Blöcke zum Aufbau von Pipelines und Datenverarbeitungsnetzwerken. Die wichtigsten Blöcke:

• BufferBlock<T> — einfache Warteschlange.
• ActionBlock<T> — Block, der eine Aktion für jedes Element ausführt.
• TransformBlock<TInput, TOutput> — transformiert jedes Element.
• BatchBlock<T> — gruppiert Elemente in Batches.

Beispiel: Pipeline mit ActionBlock

Wir setzen dieselbe Aufgabe um: Der Produzent sendet Zahlen, der Konsument verarbeitet sie.

using System.Threading.Tasks.Dataflow;

var actionBlock = new ActionBlock<int>(async item =>{    await Task.Delay(200);    Console.WriteLine($"Verarbeitet: {item * 2}");}, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 2 });

// Producerfor (int i = 0; i < 10; i++){    await actionBlock.SendAsync(i);    Console.WriteLine($"Produziert: {i}");    await Task.Delay(100);}

actionBlock.Complete();await actionBlock.Completion;Console.WriteLine("Fertig!");

Vorteile von TPL Dataflow:

• Flexible Nas