Warteschlangen und Stacks Producer-Consumer

1. Einführung

Fangen wir mit den Grundlagen an. Ein Warteschlange ist eine fundamentale Datenstruktur, die nach dem Prinzip FIFO (First-In, First-Out) funktioniert, also "wer zuerst kommt — wird zuerst bedient". Stell dir eine normale Schlange im Supermarkt vor: wer zuerst ansteht, wird zuerst bedient.

Im Multithread-Programmieren ist das Pattern Producer-Consumer eines der häufigsten und mächtigsten.

• Producer (Produzenten) — das sind Threads oder Teile der Anwendung, die Daten oder Aufgaben erzeugen und in eine gemeinsame Warteschlange legen. Sie "produzieren" Arbeit.
• Consumer (Konsumenten) — das sind Threads oder Teile der Anwendung, die Daten oder Aufgaben aus der Warteschlange nehmen und verarbeiten. Sie "konsumieren" Arbeit.

Dieses Pattern hilft, den Datenfluss zu steuern, entkoppelt Systemkomponenten (der Producer muss nicht wissen, wer und wie die Daten verarbeitet), macht das System reaktionsfähiger und hilft, die Last gleichmäßiger auf Threads zu verteilen.

Beispiel: ConcurrentQueue — Hinzufügen und Entnehmen

Schauen wir uns an, wie man Elemente zu einer ConcurrentQueue<T> hinzufügt und entnimmt.

using System.Collections.Concurrent;

ConcurrentQueue<string> tasks = new ConcurrentQueue<string>();

// Hinzufügen von Elementen (Producer)
tasks.Enqueue("Datei herunterladen");
tasks.Enqueue("Bild verarbeiten");
Console.WriteLine($"Aufgaben in der Warteschlange: {tasks.Count}"); // Ausgabe: Aufgaben in der Warteschlange: 2

// Entnehmen von Elementen (Consumer)
if (tasks.TryDequeue(out string task1))
{
    Console.WriteLine($"Aufgabe ausgeführt: {task1}"); // Ausgabe: Aufgabe ausgeführt: Datei herunterladen
}

if (tasks.TryDequeue(out string task2))
{
    Console.WriteLine($"Aufgabe ausgeführt: {task2}"); // Ausgabe: Aufgabe ausgeführt: Bild verarbeiten
}

if (!tasks.TryDequeue(out string emptyTask))
{
    Console.WriteLine("Warteschlange ist leer, keine neuen Aufgaben."); // Ausgabe: Warteschlange ist leer, keine neuen Aufgaben.
}

Grundlagen: Enqueue(), TryDequeue()

Enqueue(T item): Wird verwendet, um ein Element am Ende der Warteschlange hinzuzufügen. Diese Operation ist thread-sicher. Du kannst Enqueue gleichzeitig aus 10 verschiedenen Threads aufrufen und alle Elemente werden korrekt hinzugefügt.

TryDequeue(out T item): Wird verwendet, um zu versuchen, ein Element vom Anfang der Warteschlange zu entnehmen. Das ist die zentrale Methode für Konsumenten. Sie gibt true zurück, wenn ein Element erfolgreich entnommen wurde (der Wert landet im Out-Parameter item), und false, wenn die Warteschlange leer ist. Wichtig ist, dass TryDequeue den Thread nicht blockiert, wenn die Warteschlange leer ist.

2. Die Bedeutung von TryDequeue() und Atomarität der Operationen

Die Methode TryDequeue() ist nicht nur praktisch; sie ist kritisch wichtig für korrektes thread-sicheres Verhalten. Sie ist atomar: die Prüfung auf Leere der Warteschlange und das eigentliche Entnehmen eines Elements geschehen als eine untrennbare Operation.

Hätten wir separate Methoden IsEmpty (prüfen, ob die Warteschlange leer ist) und Dequeue (ein Element entnehmen), könnte zwischen ihren Aufrufen ein anderer Thread die Warteschlange leeren. Am Ende würde dein Dequeue eine Ausnahme werfen oder falsche Daten zurückgeben. TryDequeue verhindert genau dieses Problem.

Beispiel: Producer-Consumer mit mehreren Threads

Hier starten wir zwei Producer-Threads und einen Consumer-Thread.

using System.Collections.Concurrent;
using System.Threading;
using System.Threading.Tasks;

ConcurrentQueue<int> dataQueue = new ConcurrentQueue<int>();
bool producersDone = false; // Flag zur Signalisierung an den Consumer

void Producer(int start, int count)
{
    for (int i = 0; i < count; i++)
    {
        dataQueue.Enqueue(start + i);
        Console.WriteLine($"[P] Hinzugefügt: {start + i}");
        Thread.Sleep(10); 
    }
}

void Consumer()
{
    while (!producersDone || dataQueue.Count > 0) // Weiterlaufen, solange Daten vorhanden sind oder Producer arbeiten
    {
        if (dataQueue.TryDequeue(out int item))
        {
            Console.WriteLine($"[C] Verarbeitet: {item}");
        }
        else
        {
            Thread.Sleep(50); // Warten, wenn die Warteschlange leer ist
        }
    }
    Console.WriteLine("[C] Arbeit beendet.");
}

// Start im Main:
// Task.Run(() => Producer(1, 5));
// Task.Run(() => Producer(100, 5)); // Zweiter Producer
// Task.Run(() => Consumer());
// Thread.Sleep(600); // Zeit den Threads zu arbeiten geben
// producersDone = true; // Signalisieren, dass Producer fertig sind
// Thread.Sleep(200); // Zeit geben, dass der Consumer den Rest abholt

Beachte, dass in diesem einfachen Beispiel das Flag producersDone und Thread.Sleep zur Simulation des Beendens verwendet werden. In echten Anwendungen werden für robustere Synchronisation oft CancellationTokenSource oder BlockingCollection<T> eingesetzt.

ConcurrentQueue<T> ist ideal für Szenarien, in denen:

• die Reihenfolge der Verarbeitung wichtig ist (FIFO).
• viele Threads Elemente hinzufügen und/oder viele Threads Elemente entnehmen.
• hohe Performance ohne manuelle Lock-Verwaltung benötigt wird.

3. Stack für Producer-Consumer (LIFO)

Ein Stack ist eine andere fundamentale Datenstruktur, die nach dem Prinzip LIFO (Last-In, First-Out) funktioniert, also "wer zuletzt kommt — geht zuerst". Stell dir einen Stapel Teller vor: du nimmst immer den obersten Teller, und neue Teller legst du oben drauf.

ConcurrentStack<T> ist genauso thread-sicher wie ConcurrentQueue<T> und kann ebenfalls im Producer-Consumer-Pattern verwendet werden, aber mit umgekehrter Verarbeitungsreihenfolge.

Beispiel: ConcurrentStack — Hinzufügen und Entnehmen

using System.Collections.Concurrent;

ConcurrentStack<string> commandStack = new ConcurrentStack<string>();

// Hinzufügen von Befehlen (Producer)
commandStack.Push("Text markieren");
commandStack.Push("Schriftart ändern");
commandStack.Push("Dokument speichern");
Console.WriteLine($"Befehle im Stack: {commandStack.Count}"); // Ausgabe: Befehle im Stack: 3

// Entnehmen von Befehlen (Consumer)
if (commandStack.TryPop(out string cmd1))
{
    Console.WriteLine($"Befehl rückgängig gemacht: {cmd1}"); // Ausgabe: Befehl rückgängig gemacht: Dokument speichern
}

if (commandStack.TryPop(out string cmd2))
{
    Console.WriteLine($"Befehl rückgängig gemacht: {cmd2}"); // Ausgabe: Befehl rückgängig gemacht: Schriftart ändern
}

if (!commandStack.TryPop(out string emptyCmd))
{
    Console.WriteLine("Befehlstack ist leer."); // Ausgabe: Befehlstack ist leer.
}

4. Grundlagen: Push(), TryPop()

Push(T item): Wird verwendet, um ein Element auf die Spitze des Stacks hinzuzufügen. Die Operation ist thread-sicher.

TryPop(out T item): Wird verwendet, um zu versuchen, ein Element von der Spitze des Stacks zu entnehmen. Gibt true zurück, wenn ein Element erfolgreich entnommen wurde, und false, wenn der Stack leer ist. Wie bei TryDequeue ist dies eine atomare Operation, die Datenrennen verhindert.

Beispiel: Verwendung von ConcurrentStack für einen Objektpool

Ein Stack eignet sich gut zur Implementierung von Objektpools: nehmen — verwenden — zurückgeben.

using System.Collections.Concurrent;

class Connection { /* Einfacher Platzhalter */ public Guid Id { get; } = Guid.NewGuid(); }

ConcurrentStack<Connection> connectionPool = new ConcurrentStack<Connection>();

// Pool mit anfänglichen Verbindungen füllen
for (int i = 0; i < 3; i++)
{
    connectionPool.Push(new Connection());
}
Console.WriteLine($"Verbindungen im Pool: {connectionPool.Count}"); // Ausgabe: Verbindungen im Pool: 3

void UseConnection()
{
    if (connectionPool.TryPop(out Connection conn))
    {
        Console.WriteLine($"[Pool] Verbindung verwendet: {conn.Id}");
        // Arbeit mit der Verbindung simulieren
        Thread.Sleep(50); 
        connectionPool.Push(conn); // Zurück in den Pool
        Console.WriteLine($"[Pool] Verbindung zurückgegeben: {conn.Id}. Im Pool: {connectionPool.Count}");
    }
    else
    {
        Console.WriteLine("[Pool] Pool ist leer. Erstelle neue Verbindung.");
        // Normalerweise erstellt man hier eine neue Verbindung, wenn der Pool leer ist
        connectionPool.Push(new Connection()); 
    }
}

// Start im Main:
Task.Run(() => UseConnection());
Task.Run(() => UseConnection());
Task.Run(() => UseConnection());
Thread.Sleep(500);

In diesem Beispiel können mehrere Threads sicher Verbindungen aus dem gemeinsamen Pool entnehmen und zurückgeben.

5. Anwendungsbeispiele und Vergleich mit ConcurrentQueue

ConcurrentStack<T> wird verwendet, wenn:

• die Reihenfolge LIFO kritisch ist (z.B. Historie für Undo-Funktion).
• schneller Zugriff auf die zuletzt hinzugefügten Elemente benötigt wird (häufig bleiben sie "hot" im CPU-Cache).
• Algorithmen auf Stacks implementiert werden (Graph-DFS, Parser für Ausdrücke).

Vergleich und Auswahl der passenden Collection

ConcurrentQueue

ConcurrentStack

Die Wahl zwischen ConcurrentQueue und ConcurrentStack hängt vollständig von der benötigten Verarbeitungsreihenfolge in deinem Producer-Consumer-Szenario ab. Beide Collections bieten hohe Performance und Thread-Sicherheit "out of the box" und befreien dich von manueller Synchronisation, sodass du skalierbare Multithread-Systeme bauen kannst.

2

Aufgabe

C# SELF, Level 58, Lektion 1

Gesperrt

Producer-Consumer-Implementierung mit einer Queue

Producer-Consumer-Implementierung mit einer Queue

Öffnen