Producer-Consumer Pattern with BlockingCollection in C

This article provides an overview of the producer-consumer design pattern and demonstrates how to implement it efficiently in C# using BlockingCollection<T>. You will learn the core concepts behind the pattern, why thread synchronization is critical, and how BlockingCollection<T> eliminates the complexity of manual locking, thread signaling, and buffer management.

What is the Producer-Consumer Pattern?

The producer-consumer pattern is a classic concurrency design pattern that separates the generation of work from its execution.

• Producers are threads or tasks that generate data and place it into a shared buffer (queue).
• Consumers are threads or tasks that retrieve data from the shared buffer and process it.
• Buffer is a shared storage mechanism that temporarily holds items until consumers are ready to process them.

This separation prevents consumers from being overwhelmed by spikes in data production and ensures producers do not block while waiting for time-consuming tasks to complete.

Why Use BlockingCollection<T>?

Implementing the producer-consumer pattern manually requires handling synchronization primitives such as locks, Monitor, or SemaphoreSlim to manage thread safety, empty buffer conditions, and capacity limits.

The BlockingCollection<T> class, located in the System.Collections.Concurrent namespace, abstracts these complexities by providing:

• Thread Safety: Safe concurrent access for multiple producers and consumers without manual locking.
• Blocking Capabilities: Automatically blocks a consumer when the collection is empty, and blocks a producer when a bounded collection reaches its maximum capacity.
• Completion Signaling: Supports marking the collection as complete via CompleteAdding(), signaling consumers that no more items will arrive.
• Integration with Cancellation Tokens: Allows blocking operations to be cancelled gracefully.

Implementing Producer-Consumer with BlockingCollection<T>

Below is a complete implementation showing how to set up producers and consumers using BlockingCollection<T>.

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

class Program
{
    static async Task Main(string[] args)
    {
        // Define a bounded collection with a maximum capacity of 5 items
        using BlockingCollection<int> buffer = new BlockingCollection<int>(boundedCapacity: 5);

        // Start the producer task
        Task producer = Task.Run(() =>
        {
            for (int i = 1; i <= 10; i++)
            {
                // Add item to the collection (blocks if buffer is full)
                buffer.Add(i);
                Console.WriteLine($"Produced: {i}");
                Task.Delay(100).Wait(); // Simulate work
            }

            // Signal that no more items will be added
            buffer.CompleteAdding();
            Console.WriteLine("Producer completed adding items.");
        });

        // Start the consumer task
        Task consumer = Task.Run(() =>
        {
            // GetConsumingEnumerable blocks until items are available
            // and completes when CompleteAdding() is called and the buffer is empty
            foreach (var item in buffer.GetConsumingEnumerable())
            {
                Console.WriteLine($"Consumed: {item}");
                Task.Delay(250).Wait(); // Simulate work
            }

            Console.WriteLine("Consumer finished processing all items.");
        });

        // Wait for both tasks to complete
        await Task.WhenAll(producer, consumer);
    }
}

Key Methods Explained

• Add(T item): Adds an item to the collection. If a boundedCapacity is set and the collection is full, the calling thread blocks until space becomes available.
• CompleteAdding(): Marks the collection as closed to new additions. Any subsequent calls to Add() will throw an InvalidOperationException.
• GetConsumingEnumerable(): Returns an IEnumerable<T> that removes items from the collection as they are enumerated. It blocks when the collection is empty and terminates when CompleteAdding() has been called and all remaining items are consumed.
• Take(): Removes and returns an item from the collection. If the collection is empty, it blocks until an item is added or CompleteAdding() is invoked.

Handling Multiple Producers and Consumers

BlockingCollection<T> seamlessly scales to multiple producers and consumers without changing the core logic. You can start multiple producer tasks calling Add() and multiple consumer tasks iterating over GetConsumingEnumerable().

When working with multiple producers, ensure that CompleteAdding() is called only after all producer tasks have finished pushing data to the collection.