Pattern avanzati e caratteristiche delle collezioni Concurrent

1. «Producer‑Consumer» + Concurrent

Abbiamo già toccato il pattern «Producer‑Consumer» nella discussione su ConcurrentQueue. Vediamolo più nel dettaglio per il caso con più producer e consumer, e anche la segnalazione di termine corretta.

Il grande vantaggio di ConcurrentQueue (e di altre collezioni Concurrent) è che si occupa lei stessa della thread-safety. Non serve avvolgere Enqueue o TryDequeue in un lock — più thread possono interagire in sicurezza tramite la coda condivisa.

Esempio: Più producer e più consumer

Più thread worker generano task, e altri thread li processano.

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

ConcurrentQueue<string> taskQueue = new ConcurrentQueue<string>();
CancellationTokenSource cts = new CancellationTokenSource(); // Per cancellare il lavoro dei consumer

// Metodo per il producer
void Producer(string name, int count)
{
    for (int i = 0; i < count; i++)
    {
        string task = $"Zadacha_{name}_{i}";
        taskQueue.Enqueue(task);
        Console.WriteLine($"[P:{name}] Dobavil: {task}");
        Thread.Sleep(10); 
    }
}

// Metodo per il consumer
void Consumer(string name)
{
    while (!cts.Token.IsCancellationRequested || taskQueue.Count > 0)
    {
        if (taskQueue.TryDequeue(out string task))
        {
            Console.WriteLine($"[C:{name}] Obrabotal: {task}");
            Thread.Sleep(20); 
        }
        else
        {
            Thread.Sleep(50); // Aspettiamo se la coda è vuota
        }
    }
    Console.WriteLine($"[C:{name}] Zavershil rabotu.");
}

// Avvio dell'esempio in Main:
Task.Run(() => Producer("A", 10)); // Producer A
Task.Run(() => Producer("B", 10)); // Producer B
Task.Run(() => Consumer("1"));    // Consumer 1
Task.Run(() => Consumer("2"));    // Consumer 2

Thread.Sleep(1000); // Diamo tempo per lavorare
cts.Cancel();       // Segnale ai consumer di terminare il lavoro
Thread.Sleep(500); // Diamo tempo ai consumer di prendere il resto e terminare

Qui più producer e consumer lavorano contemporaneamente sulla stessa ConcurrentQueue senza race condition: i metodi Enqueue e TryDequeue sono atomici.

Importanza dei segnali di terminazione (CancellationTokenSource)

Usiamo CancellationTokenSource (cts) per segnalare ai consumer la necessità di terminare il lavoro. Questo è critico per il pattern Producer‑Consumer:

• I producer hanno finito. Quando l'aggiunta degli elementi è conclusa, i consumer non devono aspettare all'infinito una coda vuota.
• L'applicazione sta per chiudere. Serve fermare i consumer in modo corretto.

CancellationTokenSource e CancellationToken forniscono un meccanismo standard: il consumer controlla periodicamente IsCancellationRequested e, se necessario, chiama ThrowIfCancellationRequested().

2. BlockingCollection<T>

Anche se ConcurrentQueue<T> è adatta al pattern «producer‑consumer», richiede attese manuali quando la coda è vuota e la segnalazione di termine va fatta a mano. Per implementazioni più comode in .NET c'è BlockingCollection<T> — non è una collezione standalone, ma un wrapper su qualsiasi IProducerConsumerCollection<T> (per esempio su ConcurrentQueue).

Vantaggi di BlockingCollection:

• Operazioni bloccanti. Add()/Take() bloccano il thread se la collezione è piena/vuota. Non serve controllare manualmente IsEmpty.
• Limitazione della dimensione. Si può impostare la capacità (Capacity). Add() si bloccherà quando il limite è raggiunto — utile per controllare la memoria.
• Finalizzazione comoda. CompleteAdding() segnala la fine degli aggiungimenti, e GetConsumingEnumerable() permette al consumer di processare gli elementi fino al termine completo.

Esempio: Producer‑Consumer con BlockingCollection<T>

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

// BlockingCollection per default usa ConcurrentQueue
BlockingCollection<int> numbers = new BlockingCollection<int>(capacity: 10); // Coda con limite 10

void ProducerBC(int count)
{
    for (int i = 0; i < count; i++)
    {
        numbers.Add(i); // Blocca se la coda è piena
        Console.WriteLine($"[P] Dobavil: {i}");
        Thread.Sleep(50);
    }
    numbers.CompleteAdding(); // Segnaliamo che il producer ha finito
    Console.WriteLine("[P] Producer zavershil dobavlenie.");
}

void ConsumerBC()
{
    // GetConsumingEnumerable blocca finché ci sono elementi o finché CompleteAdding non è chiamato
    foreach (var item in numbers.GetConsumingEnumerable())
    {
        Console.WriteLine($"[C] Obrabotal: {item}");
        Thread.Sleep(100);
    }
    Console.WriteLine("[C] Consumer zavershil rabotu.");
}

// Avvio dell'esempio in Main:
Task producerTask = Task.Run(() => ProducerBC(15)); // 15 elementi, limite 10
Task consumerTask = Task.Run(ConsumerBC);
Task.WaitAll(producerTask, consumerTask); // Aspettiamo il completamento

Nota quanto il codice del consumer sia più pulito grazie a GetConsumingEnumerable(). Se ti servono operazioni bloccanti o un limite di dimensione — BlockingCollection è lo strumento giusto.

3. Metodi e proprietà aggiuntive delle collezioni Concurrent

IsEmpty, Count

• IsEmpty (bool): la collezione è vuota?
• Count (int): numero corrente di elementi.

Esempio: Uso di IsEmpty e Count

using System.Collections.Concurrent;

ConcurrentQueue<string> q = new ConcurrentQueue<string>();
Console.WriteLine($"Ochered pustaya? {q.IsEmpty}"); // True

q.Enqueue("A");
q.Enqueue("B");
Console.WriteLine($"Elementov v ocheredi: {q.Count}"); // 2
Console.WriteLine($"Ochered pustaya? {q.IsEmpty}"); // False

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

Conversione in array (ToArray())

Tutte le collezioni Concurrent forniscono il metodo ToArray(), che ritorna uno snapshot (istantanea) degli elementi.

Esempio: Uso di ToArray()

using System.Collections.Concurrent;

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

int[] items = s.ToArray(); // Crea un nuovo array: [30, 20, 10] (per lo stack LIFO)
Console.WriteLine($"Elementi nell'array: {string.Join(", ", items)}");

// La collezione rimane invariata
Console.WriteLine($"Elementi nello stack dopo ToArray: {s.Count}"); // 3

Pulizia delle collezioni

In .NET 6+ molte collezioni Concurrent hanno ottenuto il metodo Clear() per rimuovere tutti gli elementi.

Esempio: Pulizia della collezione

using System.Collections.Concurrent;

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

bag.Clear(); // Puliamo la collezione
Console.WriteLine($"Elementov v Bag posle ochistki: {bag.Count}"); // 0
Console.WriteLine($"Bag pust? {bag.IsEmpty}"); // True

4. Caratteristiche del comportamento delle collezioni Concurrent

È importante ricordare lo snapshot istantaneo dei dati. Proprietà come Count e i risultati di ToArray() riflettono lo stato della collezione in un preciso istante. In scenari con modifiche parallele questo valore può diventare obsoleto immediatamente dopo la lettura.

Esempio: Count e ToArray() — "snapshot istantaneo"

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); 
    }
}

// Avvio dell'esempio in Main:
Task.Run(AddItemsContinuously); // Thread che aggiunge continuamente elementi

Thread.Sleep(100); // Diamo un po' di tempo per l'aggiunta
Console.WriteLine($"Count corrente: {snapshotQueue.Count}"); // Può essere 50, 80, 120...
Thread.Sleep(100);
Console.WriteLine($"Count corrente di nuovo: {snapshotQueue.Count}"); // Sarà un valore diverso
int[] currentItems = snapshotQueue.ToArray();
Console.WriteLine($"Numero di elementi in ToArray(): {currentItems.Length}"); // Può differire dall'ultimo Count

Non fare affidamento su Count come garanzia rigorosa del numero attuale di elementi durante modifiche attive.

5. Sottigliezze nell'iterazione delle collezioni Concurrent

Le singole operazioni (Add, TryTake, TryPop, GetOrAdd ecc.) sono thread-safe. Ma l'iterazione con foreach su una collezione che viene modificata parallelamente da altri thread non garantisce che vedrai tutti gli elementi o solo quelli attuali — possono esserci salti e sorprese.

Esempio: Iterazione durante la modifica

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

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

// Aggiungiamo elementi iniziali
for (int i = 0; i < 10; i++) iterQueue.Enqueue(i);

// Thread-modificatore
void Modifier()
{
    for (int i = 10; i < 20; i++)
    {
        iterQueue.Enqueue(i); // Aggiungiamo nuovi elementi
        Thread.Sleep(50);
    }
}

// Thread-iteratore
void Iterator()
{
    Console.WriteLine("Nachinaem iteraciyu...");
    int count = 0;
    foreach (var item in iterQueue) // Iteriamo
    {
        Console.Write($"{item} ");
        count++;
        Thread.Sleep(30); // Simulazione di lavoro, permettiamo al modificatore di cambiare la collezione
    }
    Console.WriteLine($"\nIteraciya zavershena. Prochitano {count} elementov.");
    Console.WriteLine($"Tekuschee kolichestvo v ocheredi: {iterQueue.Count}");
}

// Avvio dell'esempio in Main:
Task.Run(Modifier);
Task.Run(Iterator);
Thread.Sleep(1500); // Diamo tempo per lavorare

Regola: se ti serve un set fisso di elementi (per esempio per un report), prendi prima uno "snapshot" con ToArray(), e poi iteralo:

// Metodo corretto per iterare se la collezione può cambiare
int[] snapshot = iterQueue.ToArray();
foreach (var item in snapshot)
{
    // Ora stai iterando su un array immutabile-snapshot
}

Questo conclude il nostro approfondimento sui pattern avanzati e le caratteristiche delle collezioni Concurrent: abbiamo visto Producer‑Consumer con più partecipanti e BlockingCollection, e abbiamo analizzato i dettagli importanti relativi a Count, ToArray() e all'iterazione.