Advanced patterns and nuances of Concurrent collections

1. "Producer‑Consumer" + Concurrent

We've already touched the Producer‑Consumer pattern when discussing ConcurrentQueue. Let's look at it in more detail for the case with multiple producers and consumers, and proper shutdown signaling.

The main beauty of ConcurrentQueue (and other Concurrent collections) is that it takes care of thread-safety for you. You don't need to wrap Enqueue or TryDequeue in a lock — multiple threads can safely interact through the shared queue.

Example: Multiple producers and multiple consumers

Several worker threads generate tasks, and several other threads process them.

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

ConcurrentQueue<string> taskQueue = new ConcurrentQueue<string>();
CancellationTokenSource cts = new CancellationTokenSource(); // For cancelling consumers

// Producer method
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); 
    }
}

// Consumer method
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); // Wait if the queue is empty
        }
    }
    Console.WriteLine($"[C:{name}] Finished.");
}

// Run example 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); // Give some time to work
cts.Cancel();       // Signal consumers to finish
Thread.Sleep(500); // Give consumers time to pick up leftovers and exit

Here multiple producers and consumers operate on a single ConcurrentQueue without data races: Enqueue and TryDequeue are atomic.

Importance of shutdown signals (CancellationTokenSource)

We use CancellationTokenSource (cts) to signal consumers to stop. This is critical for the Producer‑Consumer pattern:

• Producers finished their work. When adding elements is done, consumers shouldn't wait forever on an empty queue.
• Application is shutting down. You need to stop consumers cleanly.

CancellationTokenSource and CancellationToken provide the standard mechanism: the consumer periodically checks IsCancellationRequested and, if needed, calls ThrowIfCancellationRequested().

2. BlockingCollection<T>

Although ConcurrentQueue<T> is good for Producer‑Consumer, it requires manual waiting when the queue is empty and explicit shutdown signaling. For a more convenient implementation .NET provides BlockingCollection<T> — it's not a standalone collection but a wrapper over any IProducerConsumerCollection<T> (for example, over ConcurrentQueue).

Advantages of BlockingCollection:

• Blocking operations. Add()/Take() block the thread when the collection is full/empty. No need to manually check IsEmpty.
• Bounded capacity. You can set a capacity. Add() will block when limit is reached — handy for memory control.
• Convenient completion. CompleteAdding() signals end of adding, and GetConsumingEnumerable() lets the consumer process items until full completion.

Example: Producer‑Consumer with BlockingCollection<T>

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

// BlockingCollection uses ConcurrentQueue by default
BlockingCollection<int> numbers = new BlockingCollection<int>(capacity: 10); // Queue with limit 10

void ProducerBC(int count)
{
    for (int i = 0; i < count; i++)
    {
        numbers.Add(i); // Blocks if the collection is full
        Console.WriteLine($"[P] Added: {i}");
        Thread.Sleep(50);
    }
    numbers.CompleteAdding(); // Signal that the producer finished
    Console.WriteLine("[P] Producer finished adding.");
}

void ConsumerBC()
{
    // GetConsumingEnumerable blocks while there are items or until CompleteAdding is called
    foreach (var item in numbers.GetConsumingEnumerable())
    {
        Console.WriteLine($"[C] Processed: {item}");
        Thread.Sleep(100);
    }
    Console.WriteLine("[C] Consumer finished.");
}

// Run example in Main:
Task producerTask = Task.Run(() => ProducerBC(15)); // 15 items, limit 10
Task consumerTask = Task.Run(ConsumerBC);
Task.WaitAll(producerTask, consumerTask); // Wait for completion

Notice how much cleaner the consumer code is thanks to GetConsumingEnumerable(). If you need blocking operations or bounded capacity — BlockingCollection is your tool.

3. Additional methods and properties of Concurrent collections

IsEmpty, Count

• IsEmpty (bool): whether the collection is empty.
• Count (int): current number of elements.

Example: Using IsEmpty and Count

using System.Collections.Concurrent;

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

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

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

Converting to arrays (ToArray())

All Concurrent collections provide ToArray(), which returns a snapshot of elements.

Example: Using ToArray()

using System.Collections.Concurrent;

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

int[] items = s.ToArray(); // Creates a new array: [30, 20, 10] (for LIFO stack)
Console.WriteLine($"Items in array: {string.Join(", ", items)}");

// Collection remains unchanged
Console.WriteLine($"Items in stack after ToArray: {s.Count}"); // 3

Clearing collections

In .NET 6+ many Concurrent collections got a Clear() method to remove all elements.

Example: Clearing a collection

using System.Collections.Concurrent;

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

bag.Clear(); // Clear the collection
Console.WriteLine($"Items in Bag after clear: {bag.Count}"); // 0
Console.WriteLine($"Bag empty? {bag.IsEmpty}"); // True

4. Behavioral nuances of Concurrent collections

It's important to remember the concept of a "snapshot" of data. Properties like Count and results of ToArray() reflect the collection state at a specific moment. Under concurrent modifications that value can become outdated immediately after you read it.

Example: Count and ToArray() — a "snapshot"

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

// Run example in Main:
Task.Run(AddItemsContinuously); // Thread that keeps adding items

Thread.Sleep(100); // Give some time for adding
Console.WriteLine($"Current Count: {snapshotQueue.Count}"); // Could be 50, 80, 120...
Thread.Sleep(100);
Console.WriteLine($"Current Count again: {snapshotQueue.Count}"); // Will be different
int[] currentItems = snapshotQueue.ToArray();
Console.WriteLine($"Number of items in ToArray(): {currentItems.Length}"); // May differ from last Count

Don't rely on Count as a strict guarantee of the exact number of elements during active changes.

5. Nuances of iterating over Concurrent collections

Individual operations (Add, TryTake, TryPop, GetOrAdd, etc.) are thread-safe. But iterating with foreach over a collection that is being modified by other threads does not guarantee you'll see all items or only them — you can miss items or see surprises.

Example: Iteration during modification

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

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

// Add initial items
for (int i = 0; i < 10; i++) iterQueue.Enqueue(i);

// Modifier thread
void Modifier()
{
    for (int i = 10; i < 20; i++)
    {
        iterQueue.Enqueue(i); // Add new items
        Thread.Sleep(50);
    }
}

// Iterator thread
void Iterator()
{
    Console.WriteLine("Starting iteration...");
    int count = 0;
    foreach (var item in iterQueue) // Iterate
    {
        Console.Write($"{item} ");
        count++;
        Thread.Sleep(30); // Simulate work, allow modifier to change the collection
    }
    Console.WriteLine($"\nIteration finished. Read {count} items.");
    Console.WriteLine($"Current count in queue: {iterQueue.Count}");
}

// Run example in Main:
Task.Run(Modifier);
Task.Run(Iterator);
Thread.Sleep(1500); // Give time to work

Rule: if you need a fixed set of items (for example, for a report), first take a snapshot via ToArray(), then iterate over it:

// The correct way to iterate if the collection may change
int[] snapshot = iterQueue.ToArray();
foreach (var item in snapshot)
{
    // Now you're iterating over an immutable array-snapshot
}

This concludes our deep dive into advanced patterns and nuances of Concurrent collections: we looked at Producer‑Consumer with multiple participants and BlockingCollection, and covered important subtleties of Count, ToArray() and iteration.