c# - 生产者－消费者队列设计有什么问题？

Question

我从C#代码示例here开始。我尝试修改它的原因有两个:1)在我的场景中，所有任务都将在消费者启动之前预先放入队列中； 2)我想将worker抽象为一个单独的类，而不是将其抽象为一个单独的类Thread类中的原始WorkerQueue成员。

我的队列似乎并没有自动处理，只是挂起了，当我在Visual Studio中中断时，它停留在_th.Join()＃1的WorkerThread行上。此外，还有更好的方法来组织此 Activity 吗？关于公开WaitOne()和Join()方法的某些事情似乎是错误的，但是我想不出让WorkerThread与队列交互的适当方法。

另外，顺便说一句-如果我在q.Start(#)块的顶部调用using，则每次踢入仅一些线程(例如，线程1、2和8处理每个任务)。为什么是这样？是某种竞赛条件，还是我做错了什么？

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

namespace QueueTest
{
    class Program
    {
        static void Main(string[] args)
        {
            using (WorkQueue q = new WorkQueue())
            {
                q.Finished += new Action(delegate { Console.WriteLine("All jobs finished"); });

                Random r = new Random();
                foreach (int i in Enumerable.Range(1, 10))
                    q.Enqueue(r.Next(100, 500));

                Console.WriteLine("All jobs queued");
                q.Start(8);
            }
        }
    }

    class WorkQueue : IDisposable
    {
        private Queue<int> _jobs = new Queue<int>();
        private int _job_count;
        private EventWaitHandle _wh = new AutoResetEvent(false);
        private object _lock = new object();
        private List<WorkerThread> _th;
        public event Action Finished;

        public WorkQueue()
        {
        }

        public void Start(int num_threads)
        {
            _job_count = _jobs.Count;
            _th = new List<WorkerThread>(num_threads);
            foreach (int i in Enumerable.Range(1, num_threads))
            {
                _th.Add(new WorkerThread(i, this));
                _th[_th.Count - 1].JobFinished += new Action<int>(WorkQueue_JobFinished);
            }
        }

        void WorkQueue_JobFinished(int obj)
        {
            lock (_lock)
            {
                _job_count--;
                if (_job_count == 0 && Finished != null)
                    Finished();
            }
        }

        public void Enqueue(int job)
        {
            lock (_lock)
                _jobs.Enqueue(job);

            _wh.Set();
        }

        public void Dispose()
        {
            Enqueue(Int32.MinValue);
            _th.ForEach(th => th.Join());
            _wh.Close();
        }

        public int GetNextJob()
        {
            lock (_lock)
            {
                if (_jobs.Count > 0)
                    return _jobs.Dequeue();
                else
                    return Int32.MinValue;
            }
        }

        public void WaitOne()
        {
            _wh.WaitOne();
        }
    }

    class WorkerThread
    {
        private Thread _th;
        private WorkQueue _q;
        private int _i;

        public event Action<int> JobFinished;

        public WorkerThread(int i, WorkQueue q)
        {
            _i = i;
            _q = q;
            _th = new Thread(DoWork);
            _th.Start();
        }

        public void Join()
        {
            _th.Join();
        }

        private void DoWork()
        {
            while (true)
            {
                int job = _q.GetNextJob();
                if (job != Int32.MinValue)
                {
                    Console.WriteLine("Thread {0} Got job {1}", _i, job);
                    Thread.Sleep(job * 10); // in reality would to actual work here
                    if (JobFinished != null)
                        JobFinished(job);
                }
                else
                {
                    Console.WriteLine("Thread {0} no job available", _i);
                    _q.WaitOne();
                }
            }
        }
    }
}

Answer 1

工作线程全部在DoWork()中的_q.WaitOne()调用上阻塞。调用线程的Join()方法将导致死锁，线程永不退出。您需要添加一种机制以向工作线程发出信号以退出。在Worker中使用WaitAny测试的ManualResetEvent将完成工作。

调试技巧之一:熟悉Debug + Windows + Threads窗口。它使您可以在线程之间切换并查看它们的调用堆栈。您自己会很快发现此问题的。