为什么这个 C# 代码会抛出 SemaphoreFullException？

Question

我有以下代码抛出SemaphoreFullException，我不明白为什么？

如果我将_semaphore = new SemaphoreSlim(0, 2) 更改为

 _semaphore = new SemaphoreSlim(0, int.MaxValue)

然后一切正常。 任何人都可以找出这段代码的错误并向我解释。

 class BlockingQueue<T>
    {
        private Queue<T> _queue = new Queue<T>();
        private SemaphoreSlim _semaphore = new SemaphoreSlim(0, 2);
        public void Enqueue(T data)
        {
            if (data == null) throw new ArgumentNullException("data");
            lock (_queue)
            {
                _queue.Enqueue(data);
            }
            _semaphore.Release();
        }

        public T Dequeue()
        {
            _semaphore.Wait();
            lock (_queue)
            {
                return _queue.Dequeue();
            }
        }
    }

    public class Test
    {
        private static BlockingQueue<string> _bq = new BlockingQueue<string>();
        public static void Main()
        {
            for (int i = 0; i < 100; i++)
            {
                _bq.Enqueue("item-" + i);
            }

            for (int i = 0; i < 5; i++)
            {
                Thread t = new Thread(Produce);
                t.Start();
            }

            for (int i = 0; i < 100; i++)
            {
                Thread t = new Thread(Consume);
                t.Start();
            }
            Console.ReadLine();
        }

        private static Random _random = new Random();
        private static void Produce()
        {
            while (true)
            {
                _bq.Enqueue("item-" + _random.Next());
                Thread.Sleep(2000);
            }

        }

        private static void Consume()
        {
            while (true)
            {
                Console.WriteLine("Consumed-" + _bq.Dequeue());
                Thread.Sleep(1000);
            }

        }
    }

Answer 1

如果你想用信号量来控制并发线程的数量，那你就用错了。您应该在出队时获取信号量，并在线程完成处理该项目时释放信号量。

您现在拥有的是一个系统，它在任何时候都只允许两个项目在队列中。最初，您的信号量的计数为 2。每次您将一个项目排入队列时，计数都会减少。两个项目后，计数为 0，如果您再次尝试释放，您将获得信号量完整异常。

如果你真的想用信号量来做这件事，你需要从Enqueue 方法中删除Release 调用。并将Release 方法添加到BlockingQueue 类。然后你会写：

    private static void Consume()
    {
        while (true)
        {
            Console.WriteLine("Consumed-" + _bq.Dequeue());
            Thread.Sleep(1000);
            bq.Release();
        }

    }

这将使您的代码工作，但这不是一个很好的解决方案。一个更好的解决方案是使用BlockingCollection<T> 和两个持久的消费者。比如：

private BlockingCollection<int> bq = new BlockingCollection<int>();

void Test()
{
    // create two consumers
    var c1 = new Thread(Consume);
    var c2 = new Thread(Consume);
    c1.Start();
    c2.Start();
    // produce
    for (var i = 0; i < 100; ++i)
    {
        bq.Add(i);
    }
    bq.CompleteAdding();

    c1.Join();
    c2.Join();
}

void Consume()
{
    foreach (var i in bq.GetConsumingEnumerable())
    {
        Console.WriteLine("Consumed-" + i);
        Thread.Sleep(1000);
    }
}

这为您提供了两个持续使用这些项目的线程。好处是您可以避免为每个项目启动一个新线程（或让 RTL 分配一个池线程）的成本。相反，线程在队列上进行非忙等待。您也不必担心显式锁定等问题。代码更简单、更健壮，并且不太可能包含错误。