Usare ForEach per rimuovere elementi in un oggetto BlockingCollection

Oltre a estrarre elementi da un oggetto BlockingCollection<T> usando il metodo Take e TryTake, è anche possibile usare foreach (For Each in Visual Basic) con BlockingCollection<T>.GetConsumingEnumerable per rimuovere gli elementi finché l'aggiunta viene completata e la raccolta è vuota. Questa operazione viene definita enumerazione mutante o enumerazione di consumo perché, a differenza di un ciclo foreach tipico (For Each), questo enumeratore modifica la raccolta di origine rimuovendo elementi.

Esempio

L'esempio seguente illustra come rimuovere tutti gli elementi di un oggetto BlockingCollection<T> usando un ciclo foreach (For Each).

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

class Example
{
    // Limit the collection size to 2000 items at any given time.
    // Set itemsToProduce to > 500 to hit the limit.
    const int UpperLimit = 1000;

    // Adjust this number to see how it impacts the producing-consuming pattern.
    const int ItemsToProduce = 100;

    static readonly BlockingCollection<long> Collection =
        new BlockingCollection<long>(UpperLimit);

    // Variables for diagnostic output only.
    static readonly Stopwatch Stopwatch = new Stopwatch();
    static int TotalAdditions = 0;

    static async Task Main()
    {
        Stopwatch.Start();

        // Queue the consumer task.
        var consumerTask = Task.Run(() => RunConsumer());

        // Queue the producer tasks.
        var produceTaskOne = Task.Run(() => RunProducer("A", 0));
        var produceTaskTwo = Task.Run(() => RunProducer("B", ItemsToProduce));
        var producerTasks = new[] { produceTaskOne , produceTaskTwo };

        // Create a cleanup task that will call CompleteAdding after
        // all producers are done adding items.
        var cleanupTask = Task.Factory.ContinueWhenAll(producerTasks, _ => Collection.CompleteAdding());

        // Wait for all tasks to complete
        await Task.WhenAll(consumerTask, produceTaskOne, produceTaskTwo, cleanupTask);

        // Keep the console window open while the
        // consumer thread completes its output.
        Console.WriteLine("Press any key to exit");
        Console.ReadKey(true);
    }

    static void RunProducer(string id, int start)
    {
        var additions = 0;
        for (var i = start; i < start + ItemsToProduce; i++)
        {
            // The data that is added to the collection.
            var ticks = Stopwatch.ElapsedTicks;

            // Display additions and subtractions.
            Console.WriteLine($"{id} adding tick value {ticks}. item# {i}");

            if (!Collection.IsAddingCompleted)
            {
                Collection.Add(ticks);
            }

            // Counter for demonstration purposes only.
            additions++;

            // Comment this line to speed up the producer threads.
            Thread.SpinWait(100000);
        }

        Interlocked.Add(ref TotalAdditions, additions);
        Console.WriteLine($"{id} is done adding: {additions} items");
    }

    static void RunConsumer()
    {
        // GetConsumingEnumerable returns the enumerator for the underlying collection.
        var subtractions = 0;
        foreach (var item in Collection.GetConsumingEnumerable())
        {
            Console.WriteLine(
                $"Consuming tick value {item:D18} : item# {subtractions++} : current count = {Collection.Count}");
        }

        Console.WriteLine(
            $"Total added: {TotalAdditions} Total consumed: {subtractions} Current count: {Collection.Count}");

        Stopwatch.Stop();
    }
}

Questo esempio usa un ciclo foreach con il metodo BlockingCollection<T>.GetConsumingEnumerable nel thread di consumo, che rimuove ogni elemento dalla raccolta mentre viene enumerato. System.Collections.Concurrent.BlockingCollection<T> limita il numero massimo di elementi presenti nella raccolta in qualsiasi momento. Tale enumerazione della raccolta blocca il thread consumer se non sono disponibili elementi o se la raccolta è vuota. In questo esempio il blocco non rappresenta un problema perché il thread producer aggiunge elementi più velocemente di quanto possano essere usati.

BlockingCollection<T>.GetConsumingEnumerable restituisce un IEnumerable<T>, pertanto l'ordine non può essere garantito. Tuttavia, internamente viene usato un System.Collections.Concurrent.ConcurrentQueue<T> come tipo di raccolta sottostante, che rimuoverà dalla coda gli oggetti in seguito all'ordinamento FIFO (first-in-first-out). Se vengono effettuate chiamate simultanee a BlockingCollection<T>.GetConsumingEnumerable, queste ultime competeranno. Un elemento utilizzato (rimosso dalla coda) in un'enumerazione non può essere osservato nell'altra.

Per enumerare la raccolta senza modificarla, usare foreach (For Each) senza il metodo GetConsumingEnumerable. Tuttavia, è importante tenere presente che questo tipo di enumerazione rappresenta uno snapshot della raccolta in un momento preciso. Se altri thread aggiungono o rimuovono elementi contemporaneamente mentre si esegue il ciclo, il ciclo potrebbe non rappresentare lo stato effettivo della raccolta.

Vedi anche

• System.Collections.Concurrent
• Programmazione parallela