How to Implement a Thread-Safe Blocking Queue in .NET?

Building a Thread-Safe Blocking Queue in .NET

Multithreaded applications often require a shared queue where multiple threads add and retrieve items concurrently. A blocking queue is ideal for this, pausing adding threads when the queue is full and pausing removing threads when the queue is empty.

A Robust Blocking Queue Solution

The following improved implementation addresses potential race conditions using Monitor for synchronization and offers a graceful closure mechanism:

SizeQueue<T> Implementation

This example utilizes a synchronized Queue<T> and Monitor for efficient thread blocking and unblocking:

class SizeQueue<T>
{
    private readonly Queue<T> queue = new Queue<T>();
    private readonly int maxSize;
    private bool closing = false; // Flag for graceful closure

    public SizeQueue(int maxSize) { this.maxSize = maxSize; }

    public void Enqueue(T item)
    {
        lock (queue)
        {
            while (queue.Count >= maxSize && !closing) // Check for closure
            {
                Monitor.Wait(queue);
            }
            if (!closing) // Add only if not closing
            {
                queue.Enqueue(item);
                Monitor.PulseAll(queue); // Signal potential dequeue operations
            }
        }
    }

    public T Dequeue()
    {
        lock (queue)
        {
            while (queue.Count == 0 && !closing) // Check for closure
            {
                Monitor.Wait(queue);
            }
            if (!closing) // Dequeue only if not closing
            {
                T item = queue.Dequeue();
                Monitor.PulseAll(queue); // Signal potential enqueue operations
                return item;
            }
            else
            {
                throw new InvalidOperationException("Queue is closed.");
            }
        }
    }

    public bool TryDequeue(out T value)
    {
        lock (queue)
        {
            while (queue.Count == 0)
            {
                if (closing)
                {
                    value = default(T);
                    return false;
                }
                Monitor.Wait(queue);
            }
            value = queue.Dequeue();
            Monitor.PulseAll(queue);
            return true;
        }
    }


    public void Close()
    {
        lock (queue)
        {
            closing = true;
            Monitor.PulseAll(queue); // Wake up any waiting threads
        }
    }
}

This enhanced SizeQueue<T> provides a thread-safe, blocking queue with a Close() method for controlled shutdown, preventing deadlocks and ensuring graceful termination of consuming threads. The TryDequeue method allows for checking the queue's state before attempting to dequeue.

The above is the detailed content of How to Implement a Thread-Safe Blocking Queue in .NET?. For more information, please follow other related articles on the PHP Chinese website!