32

我最近遇到了一个生产者/消费者模式 c# 实现。它非常简单并且（至少对我而言）非常优雅。

它似乎是在 2006 年左右设计的，所以我想知道这个实现是否
- 安全
- 仍然适用

代码如下（原始代码在http://bytes.com/topic/net/answers/575276-producer-consumer#post2251375引用）

using System;  
using System.Collections;  
using System.Threading;

public class Test
{  
    static ProducerConsumer queue;

    static void Main()
    {
        queue = new ProducerConsumer();
        new Thread(new ThreadStart(ConsumerJob)).Start();

        Random rng = new Random(0);
        for (int i=0; i < 10; i++)
        {
            Console.WriteLine ("Producing {0}", i);
            queue.Produce(i);
            Thread.Sleep(rng.Next(1000));
        }
    }

    static void ConsumerJob()
    {
        // Make sure we get a different random seed from the
        // first thread
        Random rng = new Random(1);
        // We happen to know we've only got 10 
        // items to receive
        for (int i=0; i < 10; i++)
        {
            object o = queue.Consume();
            Console.WriteLine ("\t\t\t\tConsuming {0}", o);
            Thread.Sleep(rng.Next(1000));
        }
    }
}

public class ProducerConsumer
{
    readonly object listLock = new object();
    Queue queue = new Queue();

    public void Produce(object o)
    {
        lock (listLock)
        {
            queue.Enqueue(o);

            // We always need to pulse, even if the queue wasn't
            // empty before. Otherwise, if we add several items
            // in quick succession, we may only pulse once, waking
            // a single thread up, even if there are multiple threads
            // waiting for items.            
            Monitor.Pulse(listLock);
        }
    }

    public object Consume()
    {
        lock (listLock)
        {
            // If the queue is empty, wait for an item to be added
            // Note that this is a while loop, as we may be pulsed
            // but not wake up before another thread has come in and
            // consumed the newly added object. In that case, we'll
            // have to wait for another pulse.
            while (queue.Count==0)
            {
                // This releases listLock, only reacquiring it
                // after being woken up by a call to Pulse
                Monitor.Wait(listLock);
            }
            return queue.Dequeue();
        }
    }
}
4

5 回答 5

38

代码比那更旧——我在 .NET 2.0 出现之前写了一段时间。生产者/消费者队列的概念比这要早得多:)

是的，据我所知，该代码是安全的 - 但它有一些缺陷：

  • 它是非通用的。现代版本肯定是通用的。
  • 它无法停止队列。停止队列（以便所有消费者线程退出）的一种简单方法是拥有一个可以放入队列的“停止工作”令牌。然后，您可以添加与线程一样多的令牌。或者，您有一个单独的标志来指示您要停止。（这允许其他线程在完成队列中的所有当前工作之前停止。）
  • 如果工作量很小，一次只做一个工作可能不是最有效的做法。

老实说，代码背后的想法比代码本身更重要。

于 2009-11-01T08:04:33.070 回答
32

您可以执行类似以下代码片段的操作。它是通用的，并且有一种将空值（或您想使用的任何标志）排入队列的方法来告诉工作线程退出。

代码取自这里：http ://www.albahari.com/threading/part4.aspx#_Wait_and_Pulse

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;

namespace ConsoleApplication1
{

    public class TaskQueue<T> : IDisposable where T : class
    {
        object locker = new object();
        Thread[] workers;
        Queue<T> taskQ = new Queue<T>();

        public TaskQueue(int workerCount)
        {
            workers = new Thread[workerCount];

            // Create and start a separate thread for each worker
            for (int i = 0; i < workerCount; i++)
                (workers[i] = new Thread(Consume)).Start();
        }

        public void Dispose()
        {
            // Enqueue one null task per worker to make each exit.
            foreach (Thread worker in workers) EnqueueTask(null);
            foreach (Thread worker in workers) worker.Join();
        }

        public void EnqueueTask(T task)
        {
            lock (locker)
            {
                taskQ.Enqueue(task);
                Monitor.PulseAll(locker);
            }
        }

        void Consume()
        {
            while (true)
            {
                T task;
                lock (locker)
                {
                    while (taskQ.Count == 0) Monitor.Wait(locker);
                    task = taskQ.Dequeue();
                }
                if (task == null) return;         // This signals our exit
                Console.Write(task);
                Thread.Sleep(1000);              // Simulate time-consuming task
            }
        }
    }
}
于 2009-11-01T08:29:36.730 回答
23

回到那天，我从上面的代码和它来自的文章系列中了解了 Monitor.Wait/Pulse 的工作原理（以及很多关于线程的一般知识） 。所以正如 Jon 所说，它对它有很大的价值，并且确实是安全和适用的。

但是，从 .NET 4 开始，框架中有一个生产者-消费者队列实现。我只是自己发现了它，但到目前为止，它可以满足我的一切需求。

于 2012-05-02T14:10:43.763 回答
0

警告：如果您阅读评论，您会明白我的回答是错误的:)

您的代码中可能存在死锁。

想象一下以下情况，为了清楚起见，我使用了单线程方法，但应该很容易转换为带睡眠的多线程：

// We create some actions...
object locker = new object();

Action action1 = () => {
    lock (locker)
    {
        System.Threading.Monitor.Wait(locker);
        Console.WriteLine("This is action1");
    }
};

Action action2 = () => {
    lock (locker)
    {
        System.Threading.Monitor.Wait(locker);
        Console.WriteLine("This is action2");
    }
};

// ... (stuff happens, etc.)

// Imagine both actions were running
// and there's 0 items in the queue

// And now the producer kicks in...
lock (locker)
{
    // This would add a job to the queue

    Console.WriteLine("Pulse now!");
    System.Threading.Monitor.Pulse(locker);
}

// ... (more stuff)
// and the actions finish now!

Console.WriteLine("Consume action!");
action1(); // Oops... they're locked...
action2();

如果这没有任何意义，请告诉我。

如果这一点得到证实，那么您的问题的答案是，“不，这不安全”；）我希望这会有所帮助。

于 2012-01-22T00:01:37.533 回答
0 
public class ProducerConsumerProblem
    {
        private int n;
        object obj = new object();
        public ProducerConsumerProblem(int n)
        {
            this.n = n;
        }

        public void Producer()
        {

            for (int i = 0; i < n; i++)
            {
                lock (obj)
                {
                    Console.Write("Producer =>");
                    System.Threading.Monitor.Pulse(obj);
                    System.Threading.Thread.Sleep(1);
                    System.Threading.Monitor.Wait(obj);
                }
            }
        }

        public void Consumer()
        {
            lock (obj)
            {
                for (int i = 0; i < n; i++)
                {
                    System.Threading.Monitor.Wait(obj, 10);
                    Console.Write("<= Consumer");
                    System.Threading.Monitor.Pulse(obj);
                    Console.WriteLine();
                }
            }
        }
    }

    public class Program
    {
        static void Main(string[] args)
        {
            ProducerConsumerProblem f = new ProducerConsumerProblem(10);
            System.Threading.Thread t1 = new System.Threading.Thread(() => f.Producer());
            System.Threading.Thread t2 = new System.Threading.Thread(() => f.Consumer());
            t1.IsBackground = true;
            t2.IsBackground = true;
            t1.Start();
            t2.Start();
            Console.ReadLine();
        }
    }

输出

Producer =><= Consumer
Producer =><= Consumer
Producer =><= Consumer
Producer =><= Consumer
Producer =><= Consumer
Producer =><= Consumer
Producer =><= Consumer
Producer =><= Consumer
Producer =><= Consumer
Producer =><= Consumer
于 2020-03-10T16:35:17.480 回答