Einführung in Concurrent-Kollektionen

1. Historie des Problems

In einer Single-Thread-Anwendung verhalten sich Kollektionen wie List<T>, Dictionary<T> vorhersehbar. Sobald jedoch mehrere Threads gleichzeitig auf dieselbe Kollektion zugreifen, tritt das bekannte Problem auf: race conditions.

Wenn mehrere Threads versuchen, ohne ausreichende Synchronisation dieselbe Kollektion zu lesen und/oder zu schreiben, kann Folgendes passieren:

• Inkonsistente Daten: ein Element wurde von einem Thread gelöscht, während ein anderer versucht hat, es zu aktualisieren.
• Datenverlust: ein Thread hat ein Element hinzugefügt, ein anderer hat es überschrieben, ohne von der vorherigen Schreiboperation zu wissen.
• Ausnahmen: die Kollektion kann in einem ungültigen Zustand landen und du bekommst eine InvalidOperationException (z.B. "Collection was modified; enumeration operation may not execute.") oder sogar eine NullReferenceException.

Beispiel 1: Race Condition in List<T> (einfaches Addieren)

Zwei Threads inkrementieren gleichzeitig dasselbe Element in der Liste.

using System.Collections.Generic;
using System.Threading.Tasks; // Für Task.Run

class RaceConditionExample
{
    static List<int> numbers = new List<int> { 0 }; // Liste mit einem Element

    static void Main(string[] args)
    {
        Console.WriteLine("Anfangswert: " + numbers[0]); // 0

        // Starte zwei Tasks, jeder inkrementiert numbers[0]
        Task task1 = Task.Run(() => IncrementNumbers(500_000));
        Task task2 = Task.Run(() => IncrementNumbers(500_000));

        Task.WaitAll(task1, task2); // Warten auf beide Tasks

        Console.WriteLine("Endwert: " + numbers[0]); // Erwartet 1_000_000, aber...
        // Das Ergebnis ist fast immer kleiner als 1_000_000!
    }

    static void IncrementNumbers(int count)
    {
        for (int i = 0; i < count; i++)
        {
            // Diese Operation "numbers[0]++" besteht eigentlich aus 3 Schritten:
            // 1. numbers[0] lesen
            // 2. Wert um 1 erhöhen
            // 3. neuen Wert zurückschreiben in numbers[0]
            numbers[0]++; 
        }
    }
}

Warum ist das eine Race Condition? Wenn Thread A numbers[0] liest (Wert 0) und bevor A 1 schreibt, liest Thread B ebenfalls numbers[0] (auch 0), dann erhöhen beide Threads 0 auf 1 und schreiben 1. Ein Inkrement geht verloren. Die Operation numbers[0]++ ist nicht atomar.

Beispiel 2: InvalidOperationException beim Ändern eines Dictionary

Ein Thread iteriert über das Dictionary, ein anderer verändert es.

using System.Collections.Generic;
using System.Threading; // Für Thread.Sleep

class DictionaryRaceExample
{
    static Dictionary<int, string> users = new Dictionary<int, string>();

    static void Main(string[] args)
    {
        // Initialisierung des Dictionary
        for (int i = 0; i < 5; i++) users.Add(i, $"User {i}");

        // Lese-Thread
        Thread readerThread = new Thread(() =>
        {
            try
            {
                foreach (var user in users) // Iteration über das Dictionary
                {
                    Console.WriteLine($"Leser: {user.Key} - {user.Value}");
                    Thread.Sleep(10); // Arbeit simulieren
                }
            }
            catch (InvalidOperationException ex)
            {
                Console.WriteLine($"Leser: FEHLER! {ex.Message}");
            }
        });

        // Schreib-Thread
        Thread writerThread = new Thread(() =>
        {
            Thread.Sleep(5); // Dem Leser etwas Zeit geben zu starten
            for (int i = 5; i < 10; i++)
            {
                users.Add(i, $"Neuer User {i}"); // Elemente hinzufügen
                Console.WriteLine($"Schreiber: Hat User {i} hinzugefügt");
                Thread.Sleep(15);
            }
        });

        readerThread.Start();
        writerThread.Start();

        readerThread.Join(); // Auf Threads warten
        writerThread.Join();
        Console.WriteLine("Beispiel abgeschlossen.");
    }
}

Warum tritt der Fehler auf? Dictionary<TKey, TValue> (wie List<T>) ist nicht dafür gedacht, gleichzeitig von mehreren Threads gelesen und geschrieben zu werden, ohne Synchronisation. Wenn der Schreib-Thread die interne Struktur ändert, fährt der Lese-Thread mit der foreach-Schleife über bereits veränderte Daten fort, was zu einer InvalidOperationException führt.

2. Warum einfache Locks (lock) nicht immer optimal sind?

Die Idee, „alles mit einem lock zu umschließen“, ist simpel, hat aber Nachteile:

// Schlechter Ansatz: zu viel Sperrung
// (Nur zur Demonstration, so sollte man es nicht machen!)
static object _lock = new object();
static List<int> _sharedList = new List<int>();

void AddItem(int item)
{
    lock (_lock)
    {
        _sharedList.Add(item);
    }
}

int GetItemCount()
{
    lock (_lock)
    {
        return _sharedList.Count;
    }
}

• Performance (Bottleneck): lock sperrt den Zugriff auf die gesamte Kollektion. Bei 100 Threads warten 99 Threads auf einen, selbst wenn die Operationen nicht direkt kollidieren.
• Komplexität: du musst an jedem Ort, wo die Kollektion benutzt wird, an den lock denken. Eine vergessene Stelle — und die Race Condition ist zurück.
• Deadlocks: mehrere lock auf verschiedenen Objekten führen leicht zu einem deadlock.
• Iteratoren: foreach hilft nicht, wenn ein anderer Thread die Kollektion modifiziert.

Deswegen gibt es in .NET speziell entworfene thread-safe Kollektionen.

Atomare Operationen

Eine thread-safe Kollektion garantiert korrektes Verhalten bei gleichzeitigem Zugriff aus mehreren Threads ohne externe Locks seitens des Nutzers. Der Schlüssel sind atomare Operationen: eine Aktion wird vollständig ausgeführt oder gar nicht — andere Threads sehen keine „Halbzustände“.

• Hinzufügen, Entfernen, Lesen — verhalten sich so, als würden sie nacheinander ausgeführt.
• Intern werden Low-Level-Techniken verwendet: interlocked Operationen (Interlocked), Compare-And-Swap (CAS), feingranulare Locks — statt einer globalen Sperre für die ganze Kollektion.

3. Überblick über System.Collections.Concurrent

Der Namespace System.Collections.Concurrent stellt eine Reihe von Kollektionen bereit, die von Grund auf für Multithreading entworfen wurden. Ihre Philosophie: maximale Parallelität und minimale Sperrung.

• Performance: skalieren mit zunehmender Kernzahl.
• Einfachheit: man muss nicht für jede Operation manuell einen lock setzen.
• Weniger Fehler: viele Probleme durch manuelle Synchronisation entfallen.
• Optimiert für Konkurrenz: funktionieren effizient bei gleichzeitigen Hinzufügungen/Entfernungen.

4. Hauptklassen

ConcurrentQueue<T> (thread-safe Queue)

Prinzip: FIFO — "first in, first out". Szenarien: producer–consumer, Logging, Task-Queues.

using System.Collections.Concurrent;

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

void Producer() => messageQueue.Enqueue("Nachricht 1");

void Consumer()
{
    if (messageQueue.TryDequeue(out string message))
    {
        Console.WriteLine($"Verarbeitet: {message}");
    }
    else
    {
        Console.WriteLine("Queue ist leer.");
    }
}

ConcurrentStack<T> (thread-safe Stack)

Prinzip: LIFO — "last in, first out". Szenarien: Action-History, DFS-Traversal, Objektpools.

using System.Collections.Concurrent;

ConcurrentStack<int> historyStack = new ConcurrentStack<int>();

void PushAction(int value) => historyStack.Push(value);

void PopAction()
{
    if (historyStack.TryPop(out int action))
    {
        Console.WriteLine($"Aktion rückgängig gemacht: {action}");
    }
    else
    {
        Console.WriteLine("Stack ist leer.");
    }
}

ConcurrentBag<T> (thread-safe "Bag")

Unsortierte Kollektion, Reihenfolge ist nicht garantiert. Optimiert für das Szenario "ein Thread nimmt häufiger das zurück, was er selbst reingelegt hat". Sehr gut geeignet für Pools.

using System.Collections.Concurrent;

ConcurrentBag<System.Guid> objectPool = new ConcurrentBag<System.Guid>();

void AddObject() => objectPool.Add(System.Guid.NewGuid());

void TakeObject()
{
    if (objectPool.TryTake(out System.Guid obj))
    {
        Console.WriteLine($"Objekt entnommen: {obj}");
    }
    else
    {
        Console.WriteLine("Pool ist leer.");
    }
}

ConcurrentDictionary<TKey, TValue> (thread-safe Dictionary)

Unterstützt atomare Operationen zum Hinzufügen, Aktualisieren und Abrufen von Werten per Key. Hervorragend für Caches, Sessions, Counter.

using System.Collections.Concurrent;

ConcurrentDictionary<string, int> userScores = new ConcurrentDictionary<string, int>();

void UpdateScore(string user, int score)
{
    // Fügt atomar hinzu, falls nicht vorhanden, oder aktualisiert, falls vorhanden
    userScores.AddOrUpdate(user, score, (key, existingVal) => existingVal + score);
    Console.WriteLine($"Punktestand {user}: {userScores[user]}");
}

void GetScore(string user)
{
    if (userScores.TryGetValue(user, out int score))
    {
        Console.WriteLine($"Aktueller Punktestand {user}: {score}");
    }
    else
    {
        Console.WriteLine($"Benutzer {user} nicht gefunden.");
    }
}

5. Wann diese Kollektionen statt der normalen einsetzen?

• Die Anwendung ist multithreaded: bei nur einem Thread sind normale Kollektionen schneller (kein Overhead).
• Eine gemeinsame Kollektion wird von mehreren Threads genutzt: ein wichtiges Anzeichen für die Nutzung von System.Collections.Concurrent.
• Hohe Performance und Skalierung sind nötig: die Kollektionen sind für minimalen Wartaufwand konzipiert.
• Du willst Code vereinfachen: keine manuellen lock-Blöcke um jede Operation.
• Atomare Operationen benötigt: Hinzufügen/Entfernen/Abrufen lassen die Kollektion nicht inkonsistent zurück.

Verwende Concurrent-Kollektionen nicht, wenn:

• Die Anwendung strikt single-threaded ist.
• Du "transaktionale" Operationen über mehrere zusammenhängende Schritte brauchst (dann ist möglicherweise externe Synchronisation oder ein anderes Muster nötig).
• Strikte Entnahmereihenfolge wichtig ist, wo sie nicht garantiert wird (z.B. in ConcurrentBag<T>).