C#生产者/消费者模式

C# Producer/Consumer pattern

提问by Southsouth

I have simple one-producer/two-consumers code as follows but the output shows that only C2is consuming. Are there any bugs in my code?

我有如下简单的单生产者/双消费者代码，但输出显示只有C2消耗。我的代码中是否有任何错误？

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

The output is:

输出是：

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

回答by gary

For testing purposes, try adding a time delay inside the consumer code. It may be the case that "consumption" is so fast that one consumer thread empties the queue before the other consumer thread has a chance.

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

(edit)

（编辑）

As I suspected, adding a

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

Thread.Sleep(500);

线程.睡眠(500);

inside the consumer thread (to simulate some lengthy processing going on) results in both threads being utilized.

在消费者线程内部（以模拟正在进行的一些冗长的处理）导致两个线程都被利用。

回答by Simon Fox

Your producer only calls Monitor.PulseAll when the queue count is equal to 1 which will not be very often as nothing of substance is being done by the producer, this means that the first consume thread through the gate gets to dequeue the first item, the second consume thread will see no items in the queue and so hit the Monitor.Wait, and the Pulse won't happen again (probably until all but the last item is left) so that second thread will sit in that wait infinitely.

您的生产者仅在队列计数等于 1 时调用 Monitor.PulseAll，这不会经常发生，因为生产者没有做任何实质性的事情，这意味着通过门的第一个消费线程将第一个项目出列，第二个消耗线程将在队列中看不到任何项目，因此点击 Monitor.Wait，并且 Pulse 不会再次发生（可能直到除最后一项之外的所有项目都剩下），因此第二个线程将无限等待。

回答by Southsouth

Added Thread.Sleep(500); in Consumer.comsume

添加了 Thread.Sleep(500); 在 Consumer.com 中

then I have the following,

然后我有以下内容，

c2 Comsuming item1 c1 Comsuming item2 c2 Comsuming item3 c1 Comsuming item4 c2 Comsuming item5 c1 Comsuming item6 c2 Comsuming item7 c1 Comsuming item8 ..... the resule is not uncertain after adding Sleep.

c2 消耗品1 c1 消耗品2 c2 消耗品3 c1 消耗品4 c2 消耗品5 c1 消耗品6 c2 消耗品7 c1 消耗品8 ..... 添加睡眠后结果不确定。

回答by Ariel Popovsky

First, I can't reproduce your problem, here both threads consume some of the items. I guess your machine is faster but adding Sleep like gw suggest will solve that. What I would also suggest is that you don't try to sync the producer, I mean let it queue items as fast as it can and let the consumers sync to see who handles each item. I made a quick modification and it seems to be working fine:

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

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

You may also add the sleep to slow down the consumer loops.

您还可以添加 sleep 以减慢消费者循环。

回答by Mike Hall

I ran you code and had spurts of c1 doing the consuming and spurts of c2. You may just want to check this link from msdn: How to: Synchronize a Producer and a Consumer Thread (C# Programming Guide)

我运行了你的代码，并且有大量的 c1 执行消耗和 c2 的爆发。您可能只想查看 msdn 中的此链接：如何：同步生产者和消费者线程（C# 编程指南）

回答by Steve

I think your purpose is to have more than one consume threads working "in parallel". But your code is low efficient. The two consume threads are working sequentially essentially. The actual working code should be put outside the lock so the two consumer threads can run in real parallel. This improves runtime if you have multiple cores or even on a single core machine depending on the property of the work. Otherwise, there is actually no point to have more than one consume thread because anyway, all consume threads run in sequential.

我认为您的目的是让多个消费线程“并行”工作。但是你的代码效率很低。这两个消费线程本质上是按顺序工作的。实际的工作代码应该放在锁之外，这样两个消费者线程才能真正并行运行。如果您有多个核心，甚至在单核心机器上，这会提高运行时间，具体取决于工作的属性。否则，拥有多个消费线程实际上没有意义，因为无论如何，所有消费线程都是按顺序运行的。