ConcurrentBag und ConcurrentDictionary

1. Einführung

ConcurrentBag<T> ist eine thread-sichere, ungeordnete Collection. Ihr Hauptmerkmal und Vorteil steckt im Wort „Bag“ (Tüte), das bedeutet, dass die Reihenfolge der Elemente beim Entfernen nicht garantiert ist. Das heißt, das Element, das du bekommst, ist möglicherweise nicht das, das du basierend auf der Einfüge-Reihenfolge erwartet hättest. Dafür hat ConcurrentBag eine spezielle Optimierung, die sie in bestimmten Szenarien extrem schnell macht.

Eigenschaften von ConcurrentBag

Keine Reihenfolge: Im Gegensatz zu Queues (FIFO) oder Stacks (LIFO) verspricht ConcurrentBag nicht, dass TryTake() dir ein Element in einer bestimmten Reihenfolge im Bezug auf das Hinzufügen zurückgibt. Das ist ein wichtiger Unterschied.

Optimierung für lokalen Zugriff (Thread-Local Storage): Der Hauptgrund für ConcurrentBag ist ihre Performance in Szenarien, in denen der Thread, der ein Element hinzugefügt hat, sehr wahrscheinlich derselbe Thread ist, der es wieder entfernt.

Beispiel: ConcurrentBag — hinzufügen und entnehmen

using System.Collections.Concurrent;

ConcurrentBag<string> itemBag = new ConcurrentBag<string>();

// Hinzufügen von Elementen
itemBag.Add("Punkt A");
itemBag.Add("Punkt B");
itemBag.Add("Punkt C");

Console.WriteLine($"Elemente in der Tüte: {itemBag.Count}"); // Ausgabe: Elemente in der Tüte: 3

// Entnehmen von Elementen (Reihenfolge nicht garantiert!)
if (itemBag.TryTake(out string item1))
{
    Console.WriteLine($"Entnommen: {item1}"); // Kann "Punkt C", "Punkt B" oder "Punkt A" sein
}

if (itemBag.TryTake(out string item2))
{
    Console.WriteLine($"Entnommen: {item2}");
}

Console.WriteLine($"Verbleibende Elemente: {itemBag.Count}"); // Ausgabe: Verbleibende Elemente: 1

Du kannst diesen Code mehrfach laufen lassen und merken, dass die Reihenfolge der entnommenen Elemente variieren kann.

Methoden Add(), TryTake()

Add(T item): wird benutzt, um ein Element in die ConcurrentBag einzufügen. Die Operation ist thread-sicher.

TryTake(out T item): versucht, ein Element aus der ConcurrentBag zu entnehmen. Gibt true zurück, wenn ein Element erfolgreich entnommen wurde, und false, wenn die Tüte leer ist. Wichtig: TryTake blockiert den Thread nicht.

2. Einsatzszenarien

ConcurrentBag ist kein Ersatz für ConcurrentQueue oder ConcurrentStack. Sie glänzt in speziellen Fällen:

Objekt-/Ressourcenpools: wenn du einen Pool wiederverwendbarer Objekte hast und es wünschenswert ist, dass der Thread, der das Objekt zurückgegeben hat, es öfter wieder aufnimmt. Das reduziert Konkurrenz um globale Ressourcen.

Dynamische Aufgabenverteilung in TPL: interne Implementierungen wie in Parallel.ForEach und Parallel.For nutzen lokale Bags und einen «work-stealing»-Mechanismus, um Arbeit effizient zu verteilen.

Task-Pool mit ConcurrentBag und Lokalitäts-Optimierung

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

ConcurrentBag<string> taskPool = new ConcurrentBag<string>();

// Fülle den Pool mit Anfangsaufgaben
for (int i = 0; i < 10; i++)
{
    taskPool.Add($"Aufgabe {i}");
}

void Worker()
{
    // Jeder Thread versucht, eine Aufgabe zu nehmen
    while (taskPool.TryTake(out string task))
    {
        Console.WriteLine($"Thread {Thread.CurrentThread.ManagedThreadId}: verarbeitet {task}");
        Thread.Sleep(50); // Arbeit simulieren
    }
    Console.WriteLine($"Thread {Thread.CurrentThread.ManagedThreadId}: fertig.");
}

// Starte mehrere Worker-Threads
// Task.Run(Worker);
// Task.Run(Worker);
// Task.Run(Worker);
// Thread.Sleep(1000); // Zeit zum Ausführen geben

In diesem Beispiel erlaubt ConcurrentBag Threads, Aufgaben effizient zu nehmen und minimiert Sperren dank der internen Struktur.

Interne Mechanik

ConcurrentBag erreicht hohe Performance durch die Nutzung von thread-local Storage (TLS). Wenn ein Thread ein Element hinzufügt, landet es in einer lokalen Struktur für diesen Thread. Bei TryTake() wird zuerst die lokale Struktur geprüft; ist sie leer, erfolgt „work-stealing“ von anderen Threads oder vom globalen Pool. Das reduziert Konkurrenz und macht die Collection zur guten Wahl, wenn Lokalisierung des Zugriffs wichtig ist und die Reihenfolge unwichtig ist.

3. Thread-sicheres Dictionary

ConcurrentDictionary<TKey, TValue> ist eine der am häufigsten genutzten thread-sicheren Collections: ein hochperformantes Dictionary für sichere Add-/Read-/Update-/Remove-Operationen aus mehreren Threads.

Ein normales Dictionary<TKey, TValue> ist überhaupt nicht thread-sicher. Jede Schreiboperation (Hinzufügen/Ändern/Entfernen) oder sogar Lesen während eines Schreibens kann zu Ausnahmen (InvalidOperationException) oder Datenkorruption führen.

Beispiel: Problem mit normalem Dictionary (Wiederholung)

using System.Collections.Generic;
using System.Threading.Tasks;

Dictionary<int, int> concurrentDictProblem = new Dictionary<int, int>();

void AddToDict(int start, int count)
{
    for (int i = 0; i < count; i++)
    {
        // Versuch, gleichzeitig hinzuzufügen/zu ändern
        // Führt zu Ausnahmen oder inkorrektem Verhalten
        concurrentDictProblem[start + i] = start + i;
    }
}

//Aufruf im Main:
try
{
    Task t1 = Task.Run(() => AddToDict(0, 10000));
    Task t2 = Task.Run(() => AddToDict(5000, 10000)); // Überschneidende Keys
    Task.WaitAll(t1, t2);
    Console.WriteLine($"Elemente im (problematischen) Dictionary: {concurrentDictProblem.Count}");
}
catch (Exception ex)
{
    Console.WriteLine($"Fehler im normalen Dictionary: {ex.Message}");
}

Dieser Code wird mit hoher Wahrscheinlichkeit eine Ausnahme werfen oder hängen bleiben wegen Thread-Safety-Problemen.

4. Grundlegende Operationen

ConcurrentDictionary bietet spezialisierte atomare „check + act“-Operationen.

TryAdd(TKey key, TValue value): fügt das Key-Value-Paar atomar hinzu. Gibt true zurück, wenn der Key hinzugefügt wurde, und false, wenn der Key bereits existiert.

ConcurrentDictionary<string, int> scores = new ConcurrentDictionary<string, int>();
if (scores.TryAdd("Alice", 100))
    Console.WriteLine("Alice hinzugefügt."); // Ausgabe: Alice hinzugefügt.
if (!scores.TryAdd("Alice", 150))
    Console.WriteLine("Alice existiert bereits."); // Ausgabe: Alice existiert bereits.

TryGetValue(TKey key, out TValue value): holt atomar den Wert für einen Key.

if (scores.TryGetValue("Alice", out int aliceScore))
    Console.WriteLine($"Alice Score: {aliceScore}"); // Ausgabe: Alice Score: 100

TryUpdate(TKey key, TValue newValue, TValue comparisonValue): aktualisiert atomar den Wert nur, wenn der aktuelle gleich comparisonValue ist. Verhindert Race-Conditions.

// Aktueller Wert Alice = 100
if (scores.TryUpdate("Alice", 120, 100)) // Aktualisiert 100 zu 120
    Console.WriteLine("Alice Score auf 120 aktualisiert."); // Ausgabe: Alice Score auf 120 aktualisiert.
if (!scores.TryUpdate("Alice", 130, 100)) // Aktualisiert nicht, da aktuell 120 und nicht 100
    Console.WriteLine("Alice Score nicht aktualisiert (veraltete Daten)."); // Ausgabe: ...

TryRemove(TKey key, out TValue value): entfernt atomar ein Element per Key.

if (scores.TryRemove("Alice", out int removedScore))
    Console.WriteLine($"Alice entfernt, Score war: {removedScore}"); // Ausgabe: Alice entfernt, Score war: 120

5. Fortgeschrittene atomare Operationen

Diese beiden Methoden sind die Arbeitspferde von ConcurrentDictionary und decken viele Szenarien ab.

GetOrAdd(TKey key, TValue valueFactory(TKey key)): gibt atomar ein bestehendes Value für den Key zurück oder erstellt und fügt ein neues über eine Factory hinzu. Ideal für Caches und eindeutige Entitäten.

// Angenommen, wir cachen schwere Objekte
ConcurrentDictionary<int, HeavyObject> objectCache = new ConcurrentDictionary<int, HeavyObject>();

HeavyObject GetOrCreateHeavyObject(int id)
{
    // Wenn vorhanden — wird zurückgegeben, sonst wird es erstellt und hinzugefügt
    return objectCache.GetOrAdd(id, (key) =>
    {
        Console.WriteLine($"Erstelle neues HeavyObject für ID: {key}");
        return new HeavyObject(key); // Erstellung eines teuren Objekts simulieren
    });
}

// Im Main:
HeavyObject obj1 = GetOrCreateHeavyObject(1); // Erstellt neu
HeavyObject obj2 = GetOrCreateHeavyObject(2); // Erstellt neu
HeavyObject obj3 = GetOrCreateHeavyObject(1); // Gibt vorhandenes obj1 zurück

AddOrUpdate(TKey key, TValue addValue, Func<TKey, TValue, TValue> updateValueFactory): fügt atomar hinzu, wenn Key fehlt, oder aktualisiert das bestehende Value über eine Factory.

• addValue: Wert zum Hinzufügen, falls Key nicht gefunden wird.
• updateValueFactory: Funktion, die neuen Wert basierend auf Key und aktuellem Wert berechnet.

// Seitenaufruf-Zähler
ConcurrentDictionary<string, int> pageViews = new ConcurrentDictionary<string, int>();

void IncrementPageView(string page)
{
    pageViews.AddOrUpdate(page, 1, // Wenn neu, füge 1 hinzu
                          (key, existingVal) => existingVal + 1); // Sonst +1
    Console.WriteLine($"Seite '{page}' wurde {pageViews[page]} mal besucht.");
}

// Im Main:
IncrementPageView("Home");   // Home: 1
IncrementPageView("About");  // About: 1
IncrementPageView("Home");   // Home: 2
IncrementPageView("Home");   // Home: 3
IncrementPageView("Contact"); // Contact: 1

6. Beispiele für Caching oder State-Management

Daten-Caching: ConcurrentDictionary ist eine ausgezeichnete Wahl für in-memory Caches: GetOrAdd vermeidet wiederholte Erstellung teurer Objekte.

Verwaltung von User-Sessions: sicheres Speichern und Aktualisieren von Session-Daten aus verschiedenen Requests.

Statistik-Zählung: mit AddOrUpdate lässt sich bequem Event-/View-/Vote-Zähler inkrementieren.

Register/Service Locator: Speicherung registrierter Services oder Plugins, zugreifbar aus verschiedenen Threads.

ConcurrentDictionary<TKey, TValue> ist eine stark optimierte Collection, die Multithreaded-Entwicklung mit Dictionaries deutlich vereinfacht dank der Menge an atomaren Operationen ohne manuelle Synchronisation.