C#生产者/消费者模式

Question

我只有一个生产者/两个消费者的简单代码，如下所示，但是输出显示只有C2在消费。我的代码中有没有bug？

class Program
{
    static void Main(string[] args)
    {
        Object lockObj = new object();
        Queue<string>  queue = new Queue<string>();
        Producer p = new Producer(queue, lockObj);
        Consumer c1 = new Consumer(queue, lockObj, "c1");
        Consumer c2 = new Consumer(queue, lockObj, "c2");

        Thread t1 = new Thread(c1.consume);
        Thread t2 = new Thread(c2.consume);
        t1.Start();
        t2.Start();

        Thread t = new Thread(p.produce);
        t.Start();

        Console.ReadLine();
    } 
}
public class Producer
{
    Queue<string> queue;
    Object lockObject;
    static int seq = 0;
    public Producer(Queue<string> queue, Object lockObject)
    {
        this.queue = queue;
        this.lockObject = lockObject; 
    }

    public void produce()
    {
        while( seq++ <15) //just testinng 15 items
        {
            lock (lockObject)
            {
                string item = "item" + seq;
                queue.Enqueue(item);
                Console.WriteLine("Producing {0}", item);
                if (queue.Count == 1)
                { // first
                    Monitor.PulseAll(lockObject);
                }
            }
        }
    }

}

public class Consumer
{
    Queue<string> queue;
    Object lockObject;
    string name;
    public Consumer(Queue<string> queue, Object lockObject, string name)
    {
        this.queue = queue;
        this.lockObject = lockObject; 
        this.name = name;
    }

    public void consume()
    {
        string item;
        while (true)
        {
            lock (lockObject)
            {
                if (queue.Count == 0)
                { 
                    Monitor.Wait(lockObject);
                    continue; 
                }
                item = queue.Dequeue();
                Console.WriteLine(" {0} Consuming {1}", name, item);
            }
        }
    }
}

输出为：

Producing item1
 c2 Consuming item1

Producing item2
 c2 Consuming item2

Producing item3
 c2 Consuming item3

Producing item4
 c2 Consuming item4

Producing item5
 c2 Consuming item5

Producing item6
 c2 Consuming item6

Producing item7
 c2 Consuming item7

Producing item8
 c2 Consuming item8

Producing item9
 c2 Consuming item9

Producing item10
 c2 Consuming item10

Producing item11
 c2 Consuming item11

Producing item12
 c2 Consuming item12

Producing item13
 c2 Consuming item13

Producing item14
 c2 Consuming item14

Producing item15
 c2 Consuming item15

Answer 1

首先，我不能重现你的问题，这里两个线程都消耗了一些项。我猜你的机器更快，但是像gw建议的那样添加睡眠可以解决这个问题。我还建议你不要尝试同步生产者，我的意思是让它尽可能快地对项目进行排队，让消费者同步，看看谁在处理每个项目。我做了一个快速的修改，它似乎工作得很好：

static void Main()
    {
        Object lockObj = new object();
        Queue<string> queue = new Queue<string>();
        Producer p = new Producer(queue);
        Comsumer c1 = new Comsumer(queue, lockObj, "c1");
        Comsumer c2 = new Comsumer(queue, lockObj, "c2");

        Thread t1 = new Thread(c1.consume);
        Thread t2 = new Thread(c2.consume);
        t1.Start();
        t2.Start();

        Thread t = new Thread(p.produce);
        t.Start();

        Console.ReadLine();
    }
}
public class Producer
{
    Queue<string> queue;
    static int seq;
    public Producer(Queue<string> queue)
    {
        this.queue = queue;
    }

    public void produce()
    {
        while (seq++ < 1000) //just testinng 15 items
        {
            string item = "item" + seq;
            queue.Enqueue(item);
            Console.WriteLine("Producing {0}", item);                
        }
    }
}

public class Comsumer
{
    Queue<string> queue;
    Object lockObject;
    string name;
    public Comsumer(Queue<string> queue, Object lockObject, string name)
    {
        this.queue = queue;
        this.lockObject = lockObject;
        this.name = name;
    }

    public void consume()
    {
        string item;
        while (true)
        {
            lock (lockObject)
            {
                if (queue.Count == 0)
                {
                    continue;
                }
                item = queue.Dequeue();
                Console.WriteLine(" {0} Comsuming {1}", name, item);
            }                
        }
    }
}

您还可以添加睡眠来减慢消费者循环。

Answer 2

出于测试目的，请尝试在使用者代码中添加时间延迟。可能的情况是“消费”太快，以至于一个消费者线程在另一个消费者线程有机会之前清空队列。

(编辑)

正如我所怀疑的那样，添加一个

Thread.Sleep(500)；

在使用者线程内部(为了模拟一些冗长的处理过程)，两个线程都会被利用。

Answer 3

你的生产者只在队列计数等于1时调用Monitor.PulseAll，这不会经常发生，因为生产者并没有做任何实质性的事情，这意味着通过gate的第一个消费线程将第一个项目出队，第二个消费线程将看不到队列中的任何项目，因此命中Monitor.Wait，并且脉冲不会再次发生(可能直到最后一个项目被剩下)，因此第二个线程将无限期地等待。