Patrons avancés et particularités des collections Concurrent

1. « Producteur‑Consommateur » + Concurrent

Nous avons déjà évoqué le patron « Producteur‑Consommateur » en parlant de ConcurrentQueue. Regardons-le en détail pour le cas de plusieurs producteurs et consommateurs, ainsi que la signalisation correcte de fin.

Le grand avantage de ConcurrentQueue (et des autres collections Concurrent) est qu’elle gère la sécurité entre threads elle‑même. Pas besoin d’enrober Enqueue ou TryDequeue dans un lock — plusieurs threads peuvent interagir en toute sécurité via la file partagée.

Exemple : Plusieurs producteurs et plusieurs consommateurs

Plusieurs threads workers génèrent des tâches, et plusieurs autres threads les traitent.

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

ConcurrentQueue<string> taskQueue = new ConcurrentQueue<string>();
CancellationTokenSource cts = new CancellationTokenSource(); // Pour annuler le travail des consommateurs

// Méthode du producteur
void Producer(string name, int count)
{
    for (int i = 0; i < count; i++)
    {
        string task = $"Task_{name}_{i}";
        taskQueue.Enqueue(task);
        Console.WriteLine($"[P:{name}] Added: {task}");
        Thread.Sleep(10); 
    }
}

// Méthode du consommateur
void Consumer(string name)
{
    while (!cts.Token.IsCancellationRequested || taskQueue.Count > 0)
    {
        if (taskQueue.TryDequeue(out string task))
        {
            Console.WriteLine($"[C:{name}] Processed: {task}");
            Thread.Sleep(20); 
        }
        else
        {
            Thread.Sleep(50); // Attendre si la file est vide
        }
    }
    Console.WriteLine($"[C:{name}] Finished work.");
}

// Lancement de l'exemple dans Main:
Task.Run(() => Producer("A", 10)); // Producteur A
Task.Run(() => Producer("B", 10)); // Producteur B
Task.Run(() => Consumer("1"));    // Consommateur 1
Task.Run(() => Consumer("2"));    // Consommateur 2

Thread.Sleep(1000); // Donner du temps pour travailler
cts.Cancel();       // Signaler aux consommateurs d'arrêter
Thread.Sleep(500); // Donner du temps aux consommateurs pour traiter les restes et terminer

Ici plusieurs producteurs et consommateurs travaillent simultanément sur une même ConcurrentQueue sans conditions de compétition : les méthodes Enqueue et TryDequeue sont atomiques.

Importance des signaux de fin (CancellationTokenSource)

On utilise CancellationTokenSource (cts) pour signaler aux consommateurs qu’ils doivent terminer. C’est crucial pour le patron Producer‑Consumer :

• Les producteurs ont fini leur travail. Quand l’ajout d’éléments est terminé, les consommateurs ne doivent pas attendre indéfiniment une file vide.
• L’application se termine. Il faut arrêter proprement les consommateurs.

CancellationTokenSource et CancellationToken fournissent un mécanisme standard : le consommateur vérifie périodiquement IsCancellationRequested et, si nécessaire, appelle ThrowIfCancellationRequested().

2. BlockingCollection<T>

Bien que ConcurrentQueue<T> convienne pour le pattern « producteur‑consommateur », elle nécessite une attente manuelle quand la file est vide et une signalisation manuelle de fin. Pour une implémentation plus pratique en .NET, il y a BlockingCollection<T> — ce n’est pas une collection indépendante mais un wrapper autour de n’importe quelle IProducerConsumerCollection<T> (par exemple au‑dessus de ConcurrentQueue).

Avantages de BlockingCollection :

• Opérations bloquantes. Add()/Take() bloquent le thread si la collection est pleine/vide. Pas besoin de vérifier manuellement IsEmpty.
• Limitation de taille. On peut définir une capacité (Capacity). Add() se bloquera quand la limite sera atteinte — pratique pour contrôler la mémoire.
• Finalisation pratique. CompleteAdding() signale la fin des ajouts, et GetConsumingEnumerable() permet au consommateur de traiter les éléments jusqu’à la fin complète.

Exemple : Producer‑Consumer avec BlockingCollection<T>

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

// BlockingCollection utilise par défaut ConcurrentQueue
BlockingCollection<int> numbers = new BlockingCollection<int>(capacity: 10); // File avec limite 10

void ProducerBC(int count)
{
    for (int i = 0; i < count; i++)
    {
        numbers.Add(i); // Bloque si la file est pleine
        Console.WriteLine($"[P] Added: {i}");
        Thread.Sleep(50);
    }
    numbers.CompleteAdding(); // Signalons que le producteur a fini
    Console.WriteLine("[P] Producer finished adding.");
}

void ConsumerBC()
{
    // GetConsumingEnumerable bloque tant qu'il y a des éléments ou tant que CompleteAdding n'a pas été appelé
    foreach (var item in numbers.GetConsumingEnumerable())
    {
        Console.WriteLine($"[C] Processed: {item}");
        Thread.Sleep(100);
    }
    Console.WriteLine("[C] Consumer finished work.");
}

// Lancement de l'exemple dans Main:
Task producerTask = Task.Run(() => ProducerBC(15)); // 15 éléments, limite 10
Task consumerTask = Task.Run(ConsumerBC);
Task.WaitAll(producerTask, consumerTask); // Attendre la fin

Notez combien le code du consommateur est plus propre grâce à GetConsumingEnumerable(). Si vous avez besoin d’opérations bloquantes ou d’une limite de taille — BlockingCollection est votre outil.

3. Méthodes et propriétés supplémentaires des collections Concurrent

IsEmpty, Count

• IsEmpty (bool) : la collection est‑elle vide.
• Count (int) : nombre actuel d’éléments.

Exemple : Utilisation de IsEmpty et Count

using System.Collections.Concurrent;

ConcurrentQueue<string> q = new ConcurrentQueue<string>();
Console.WriteLine($"Is the Queue' empty? {q.IsEmpty}"); // True

q.Enqueue("A");
q.Enqueue("B");
Console.WriteLine($"Elements v in queue: {q.Count}"); // 2
Console.WriteLine($"Is the Queue' empty? {q.IsEmpty}"); // False

q.TryDequeue(out var itemA);
Console.WriteLine($"Elements v in queue: {q.Count}"); // 1

Conversion en tableaux (ToArray())

Toutes les collections Concurrent fournissent la méthode ToArray(), qui retourne un instantané des éléments.

Exemple : Utilisation de ToArray()

using System.Collections.Concurrent;

ConcurrentStack<int> s = new ConcurrentStack<int>();
s.Push(10);
s.Push(20);
s.Push(30);

int[] items = s.ToArray(); // Créera un nouveau tableau : [30, 20, 10] (pour une pile LIFO)
Console.WriteLine($"Elements in array: {string.Join(", ", items)}");

// La collection reste inchangée
Console.WriteLine($"Elements v stack posle ToArray: {s.Count}"); // 3

Cleaning collections (Clear)

Dans .NET 6+ beaucoup de collections Concurrent ont obtenu la méthode Clear() pour supprimer tous les éléments.

Exemple : Nettoyage d'une collection

using System.Collections.Concurrent;

ConcurrentBag<string> bag = new ConcurrentBag<string>();
bag.Add("Alpha");
bag.Add("Beta");
Console.WriteLine($"Elements v Bag: {bag.Count}"); // 2

bag.Clear(); // On nettoie la collection
Console.WriteLine($"Elements v Bag posle cleaning: {bag.Count}"); // 0
Console.WriteLine($"Bag empty? {bag.IsEmpty}"); // True

4. Particularités du comportement des collections Concurrent

Il est important de se rappeler le concept de « snapshot » (instantané) des données. Les propriétés comme Count et les résultats de ToArray() reflètent l’état de la collection à un instant précis. En présence de modifications parallèles, cette valeur peut devenir obsolète littéralement juste après l’avoir lue.

Exemple : Count et ToArray() — « instantané »

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

ConcurrentQueue<int> snapshotQueue = new ConcurrentQueue<int>();

void AddItemsContinuously()
{
    for (int i = 0; i < 1000; i++)
    {
        snapshotQueue.Enqueue(i);
        Thread.Sleep(1); 
    }
}

// Lancement de l'exemple dans Main:
Task.Run(AddItemsContinuously); // Thread qui ajoute en continu

Thread.Sleep(100); // Donner un peu de temps pour ajouter
Console.WriteLine($"Current Count: {snapshotQueue.Count}"); // Peut être 50, 80, 120...
Thread.Sleep(100);
Console.WriteLine($"Current Count again: {snapshotQueue.Count}"); // Sera une autre valeur
int[] currentItems = snapshotQueue.ToArray();
Console.WriteLine($"Quantity elements v ToArray(): {currentItems.Length}"); // Peut différer du dernier Count

Ne vous fiez pas à Count comme garantie stricte du nombre actuel d’éléments pendant des modifications actives.

5. Nuances de l’itération sur les collections Concurrent

Les opérations individuelles (Add, TryTake, TryPop, GetOrAdd, etc.) sont thread‑safe. Mais l’itération avec foreach sur une collection qui est modifiée en parallèle par d’autres threads ne garantit pas que vous verrez tous les éléments ou uniquement eux — des omissions et des surprises sont possibles.

Exemple : Itération pendant modification

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

ConcurrentQueue<int> iterQueue = new ConcurrentQueue<int>();

// Ajout des éléments initiaux
for (int i = 0; i < 10; i++) iterQueue.Enqueue(i);

// Thread modificateur
void Modifier()
{
    for (int i = 10; i < 20; i++)
    {
        iterQueue.Enqueue(i); // On ajoute de nouveaux éléments
        Thread.Sleep(50);
    }
}

// Thread itérateur
void Iterator()
{
    Console.WriteLine("Starting iteration...");
    int count = 0;
    foreach (var item in iterQueue) // Itération
    {
        Console.Write($"{item} ");
        count++;
        Thread.Sleep(30); // Simulation de travail, permettre au modificateur de changer la collection
    }
    Console.WriteLine($"\nIteration finished. Read {count} elements.");
    Console.WriteLine($"Current kolichestvo v in queue: {iterQueue.Count}");
}

// Lancement de l'exemple dans Main:
Task.Run(Modifier);
Task.Run(Iterator);
Thread.Sleep(1500); // Donner du temps pour travailler

Règle : si vous avez besoin d’un ensemble fixe d’éléments (par ex. pour un rapport), prenez d’abord un « snapshot » via ToArray(), puis itérez dessus :

// La façon correcte d'itérer si la collection peut changer
int[] snapshot = iterQueue.ToArray();
foreach (var item in snapshot)
{
    // Maintenant vous itérez sur un tableau‑snapshot immuable
}

Ceci conclut notre plongée dans les patrons avancés et les particularités des collections Concurrent : nous avons regardé Producer‑Consumer avec plusieurs participants et BlockingCollection, et expliqué les nuances importantes liées à Count, ToArray() et l’itération.