ConcurrentBag et ConcurrentDictionary

1. Introduction

ConcurrentBag<T> est une collection thread-safe non ordonnée. Sa principale caractéristique et avantage résident dans le mot « Bag » (sac), ce qui implique que l'ordre des éléments n'est pas garanti lors de la récupération. Ça veut dire que l'élément que vous extrayez peut ne pas être celui que vous attendiez d'après l'ordre d'ajout. En revanche, ConcurrentBag dispose d'une optimisation unique qui la rend extrêmement rapide dans certains scénarios.

Caractéristiques de ConcurrentBag

Absence d'ordre : Contrairement aux queues (FIFO) et aux stacks (LIFO), ConcurrentBag ne garantit pas que TryTake() vous retournera un élément dans un ordre particulier par rapport à son ajout. C'est une différence clé.

Optimisation pour l'accès local (Thread-Local Storage) : La raison principale d'être de ConcurrentBag est ses performances dans les scénarios où le thread qui a ajouté l'élément est très probablement le même qui le récupérera.

Exemple : ConcurrentBag — ajout et extraction

using System.Collections.Concurrent;

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

// Ajout d'éléments
itemBag.Add("Point A");
itemBag.Add("Point B");
itemBag.Add("Point C");

Console.WriteLine($"Éléments dans le sac : {itemBag.Count}"); // Sortie : Éléments dans le sac : 3

// Extraction des éléments (l'ordre n'est pas garanti !)
if (itemBag.TryTake(out string item1))
{
    Console.WriteLine($"Extrait : {item1}"); // Peut être "Point C", "Point B" ou "Point A"
}

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

Console.WriteLine($"Éléments restants : {itemBag.Count}"); // Sortie : Éléments restants : 1

Vous pouvez exécuter ce code plusieurs fois et remarquer que l'ordre des éléments extraits peut varier.

Méthodes Add(), TryTake()

Add(T item) : utilisé pour ajouter un élément dans le ConcurrentBag. L'opération est thread-safe.

TryTake(out T item) : tentative d'extraction d'un élément du ConcurrentBag. Retourne true si un élément a été extrait avec succès, et false si le sac est vide. Important : TryTake ne bloque pas le thread.

2. Scénarios d'utilisation

ConcurrentBag n'est pas un remplacement de ConcurrentQueue ou ConcurrentStack. Elle brille dans des cas spécifiques :

Pools d'objets/ressources : quand vous avez un pool d'objets réutilisables, et qu'il est souhaitable que le thread qui retourne l'objet soit le plus souvent celui qui le reprendra. Ça réduit la contention sur la ressource partagée.

Répartition dynamique des tâches dans le TPL : le fonctionnement interne de constructions comme Parallel.ForEach et Parallel.For utilise des sacs locaux et un mécanisme de « work-stealing » pour répartir efficacement le travail.

Pool de tâches avec ConcurrentBag et optimisation de la localité

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

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

// On remplit le pool avec des tâches initiales
for (int i = 0; i < 10; i++)
{
    taskPool.Add($"Tâche {i}");
}

void Worker()
{
    // Chaque thread tente de prendre une tâche
    while (taskPool.TryTake(out string task))
    {
        Console.WriteLine($"Thread {Thread.CurrentThread.ManagedThreadId} : Traite {task}");
        Thread.Sleep(50); // Simulation de travail
    }
    Console.WriteLine($"Thread {Thread.CurrentThread.ManagedThreadId} : Travail terminé.");
}

// On lance plusieurs workers
// Task.Run(Worker);
// Task.Run(Worker);
// Task.Run(Worker);
// Thread.Sleep(1000); // On laisse le temps d'exécution

Dans cet exemple, ConcurrentBag permet aux threads de prendre des tâches efficacement, en minimisant les lock grâce à sa structure interne.

Mécanique interne

ConcurrentBag atteint de hautes performances en utilisant un stockage local au thread (TLS). Quand un thread ajoute un élément, il est placé dans une structure locale au thread. Lors d'un TryTake(), la structure locale est d'abord lue ; si elle est vide — on effectue du « work-stealing » depuis d'autres threads ou depuis le pool global. Ça réduit la contention et rend la collection idéale quand la localité d'accès est importante et que l'ordre ne l'est pas.

3. Dictionnaire thread-safe

ConcurrentDictionary<TKey, TValue> est l'une des collections thread-safe les plus utilisées : un dictionnaire haute performance pour des opérations sûres d'ajout, lecture, mise à jour et suppression depuis plusieurs threads.

Un Dictionary<TKey, TValue> classique n'est absolument pas thread-safe. Toute écriture (ajout/modification/suppression) ou même une lecture pendant une écriture peut conduire à des exceptions (InvalidOperationException) ou à la corruption des données.

Exemple : problème avec un Dictionary classique (rappel)

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++)
    {
        // Tentative d'ajout/modification concurrente
        // Cela va provoquer des exceptions ou un comportement incorrect
        concurrentDictProblem[start + i] = start + i;
    }
}

// Lancement de l'exemple dans Main :
try
{
    Task t1 = Task.Run(() => AddToDict(0, 10000));
    Task t2 = Task.Run(() => AddToDict(5000, 10000)); // Chevauchement des clés
    Task.WaitAll(t1, t2);
    Console.WriteLine($"Éléments dans le dictionnaire (problématique) : {concurrentDictProblem.Count}");
}
catch (Exception ex)
{
    Console.WriteLine($"Erreur dans le dictionnaire classique : {ex.Message}");
}

Ce code va presque assurément lever une exception ou bloquer à cause de problèmes de thread-safety.

4. Opérations principales

ConcurrentDictionary fournit des opérations atomiques spécialisées « check + act ».

TryAdd(TKey key, TValue value) : ajoute atomiquement une paire clé-valeur. Retourne true si la clé a été ajoutée, et false si la clé existe déjà.

ConcurrentDictionary<string, int> scores = new ConcurrentDictionary<string, int>();
if (scores.TryAdd("Alice", 100))
    Console.WriteLine("Alice ajoutée."); // Sortie : Alice ajoutée.
if (!scores.TryAdd("Alice", 150))
    Console.WriteLine("Alice existe déjà."); // Sortie : Alice existe déjà.

TryGetValue(TKey key, out TValue value) : récupère atomiquement la valeur pour une clé.

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

TryUpdate(TKey key, TValue newValue, TValue comparisonValue) : met atomiquement à jour la valeur seulement si la valeur actuelle est égale à comparisonValue. Prévient les races.

// Valeur actuelle de Alice = 100
if (scores.TryUpdate("Alice", 120, 100)) // Mettra à jour 100 en 120
    Console.WriteLine("Score d'Alice mis à jour à 120."); // Sortie : Score d'Alice mis à jour à 120.
if (!scores.TryUpdate("Alice", 130, 100)) // Ne mettra pas à jour car la valeur actuelle est 120, pas 100
    Console.WriteLine("Score d'Alice non mis à jour (données obsolètes)."); // Sortie : ...

TryRemove(TKey key, out TValue value) : supprime atomiquement l'élément par clé.

if (scores.TryRemove("Alice", out int removedScore))
    Console.WriteLine($"Alice supprimée, le score était : {removedScore}"); // Sortie : Alice supprimée, le score était : 120

5. Opérations atomiques avancées

Ces deux méthodes sont les chevaux de trait de ConcurrentDictionary, couvrant de nombreux scénarios.

GetOrAdd(TKey key, TValue valueFactory(TKey key)) : retourne atomiquement la valeur existante pour la clé ou crée et ajoute une nouvelle via une factory. Idéal pour des caches et des entités uniques.

// Supposons qu'on met en cache des objets lourds
ConcurrentDictionary<int, HeavyObject> objectCache = new ConcurrentDictionary<int, HeavyObject>();

HeavyObject GetOrCreateHeavyObject(int id)
{
    // Si existant — le retourne, sinon le crée et l'ajoute
    return objectCache.GetOrAdd(id, (key) =>
    {
        Console.WriteLine($"Création d'un nouveau HeavyObject pour ID : {key}");
        return new HeavyObject(key); // Simulation de création d'un objet coûteux
    });
}

// Dans Main :
HeavyObject obj1 = GetOrCreateHeavyObject(1); // Crée un nouveau
HeavyObject obj2 = GetOrCreateHeavyObject(2); // Crée un nouveau
HeavyObject obj3 = GetOrCreateHeavyObject(1); // Retourne l'existant obj1

AddOrUpdate(TKey key, TValue addValue, Func<TKey, TValue, TValue> updateValueFactory) : ajoute atomiquement la valeur si la clé est absente, ou met à jour l'existante via une factory.

• addValue : valeur à ajouter si la clé n'est pas trouvée.
• updateValueFactory : fonction qui calcule la nouvelle valeur à partir de la clé et de la valeur actuelle.

// Comptage des visites de page
ConcurrentDictionary<string, int> pageViews = new ConcurrentDictionary<string, int>();

void IncrementPageView(string page)
{
    pageViews.AddOrUpdate(page, 1, // Si la page est nouvelle, ajouter 1
                          (key, existingVal) => existingVal + 1); // Sinon incrémenter de 1
    Console.WriteLine($"La page '{page}' a été visitée {pageViews[page]} fois.");
}

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

6. Exemples d'utilisation pour le caching ou la gestion d'états

Mise en cache des données : ConcurrentDictionary est un excellent choix pour un cache in-memory : GetOrAdd évite la recréation d'objets coûteux.

Gestion des sessions utilisateur : stockage et mise à jour sûrs des données de session depuis différentes requêtes.

Comptage de statistiques : avec AddOrUpdate il est pratique d'incrémenter des compteurs d'événements, de vues, de votes, etc.

Registres / Service Locator : stockage des services ou plugins enregistrés, accessibles depuis différents threads.

ConcurrentDictionary<TKey, TValue> est une collection hautement optimisée, simplifiant grandement le développement multithread avec des dictionnaires grâce à son ensemble d'opérations atomiques sans synchronisation manuelle.