当前位置:  编程技术>.net/c#/asp.net

深入多线程之:深入生产者、消费者队列分析

    来源: 互联网  发布时间:2014-10-20

    本文导语:  上次我们使用AutoResetEvent实现了一个生产/消费者队列。这一次我们要使用Wait和Pulse方法来实现一个更强大的版本,它允许多个消费者,每一个消费者都在自己的线程中运行。 我们使用数组来跟踪线程。 Thread[] _workers; 通过跟踪...

上次我们使用AutoResetEvent实现了一个生产/消费者队列。这一次我们要使用Wait和Pulse方法来实现一个更强大的版本,它允许多个消费者,每一个消费者都在自己的线程中运行。

我们使用数组来跟踪线程。

Thread[] _workers;

通过跟踪线程可以让我们在所有的线程都结束后再结束我们的队列任务。

每一个消费者线程都执行一个叫做Consume的方法,在一个for循环中,我们可以创建和启动线程。例如:

代码如下:

       public PCQueue(int workerCount)
        {
            _workers = new Thread[workerCount];
            for (int i = 0; i < workerCount; i++)
                (_workers[i] = new Thread(Consume)).Start();
        }

上次我们使用的是一个字符串来代表任务,这次我们使用Action委托,它的定义如下:

Public delegate void Action();

为了表示一系列的任务,我们使用Queue 集合,例如:

Queue _itemQ = new Queue();

在我们调用生产(EnqueueItem)和消费(Consume)方法前,还是完整的看一看代码吧:

代码如下:

class PCQueue
    {
        readonly object _locker = new object();
        Thread[] _workers;
        Queue _itemQ = new Queue(); //保存任务的队列
        public PCQueue(int workerCount)
        {
            _workers = new Thread[workerCount];
            for (int i = 0; i < workerCount; i++)
                (_workers[i] = new Thread(Consume)).Start();
        }

        public void Shutdown(bool waitForWorkers)
        {
           //为每一个线程插入一个null item,可以是每一个worker 退出
            foreach (Thread worker in _workers)
                EnqueueItem(null);

           //等待所有的线程退出。
            if (waitForWorkers)
                foreach (Thread worker in _workers)
                    worker.Join();
        }

        public void EnqueueItem(Action item)
        {
            lock (_locker)
            {
                _itemQ.Enqueue(item);
                Monitor.Pulse(_locker); //通知等待队列中的线程
            }
        }

        void Consume()
        {
            while (true)
            {
                Action item;
                lock (_locker)
                {
                    while (_itemQ.Count == 0)
                    {
                        Monitor.Wait(_locker); //释放锁,并阻止当前线程,直到其他线程发送pulse信号。                    }
                    item = _itemQ.Dequeue();
                }

                if (item == null) return; //退出的信号
                item();
            }
        }
    }


我们可以有一个退出策略,插入一个null item作为consumer退出的信号。如果我们想要快速的退出,可以使用一个独立的”cancel” 标记,因为我们支持多个consumers,所以我们必须为每一个consumer插入一个null item。

下面是Main方法。使用两个consumer线程,然后让这两个consumers执行10个委托。

代码如下:

public static void Main()
        {
            PCQueue q = new PCQueue(2);
            Console.WriteLine("Enqueuing 10 items...");

            for (int i = 0; i < 10; i++)
            {
                int itemNumber = i;
                q.EnqueueItem(() =>
                    {
                        Thread.Sleep(1000); //模拟耗时的工作
                        Console.WriteLine(" Task " + itemNumber);
                    });
            }

            q.Shutdown(true); //等待关闭
            Console.WriteLine();
            Console.WriteLine("Workers complete!");
        }

下面让我们细致的看一看EnqueueItem方法:

代码如下:

public void EnqueueItem(Action item)
        {
            lock (_locker)
            {
                _itemQ.Enqueue(item);
                Monitor.Pulse(_locker); //通知等待队列中的线程
            }
        }

因为我们的队列_itemQ被多线程环境使用,因此在对_itemQ进行读取的时候需要加锁lock.

因为我们插入了一个新的任务,我们必须修改阻塞条件,也就是调用pulse方法,来唤醒调用了wait方法的线程。


出于对效率的考虑,当插入一个Item的时候使用Pulse来代替PulseAll方法,因为大部分时候每一个Item只需要一个consumer来执行。如果你有一个冰淇淋,你不可能叫30个睡眠的孩子都起来吃它,同样,对于一个item,同时唤醒30个consumers一点好处都没有。


让我们再看看Consumer方法。

我们希望当没什么事情做的时候,线程阻塞就可以了,换句话说,队列中没有item的时候,线程就应该阻塞。因此我们的阻塞条件是_itemQ.Count ==0;

代码如下:

Action item;
                lock (_locker)
                {
                    while (_itemQ.Count == 0)
                    {
                        Monitor.Wait(_locker); //释放锁,并阻止当前线程,直到其他线程发送pulse信号。                    }
                    item = _itemQ.Dequeue();
                }

                if (item == null) return; //退出的信号
                item();


while循环退出的时候也意味着_itemQ 至少有一个item。我们必须在释放锁之前调用你哦个dequeue方法来获取item,考虑下下面的代码:
代码如下:

lock (_locker)
                {
                    while (_itemQ.Count == 0)
                    {
                        Monitor.Wait(_locker); //释放锁,并阻止当前线程,直到其他线程发送pulse信号。                    }
                }
                //现在在这里可能被抢占,_itemQ可能被修改
                lock (_locker)
                {
                    item = _itemQ.Dequeue();
                }

在item被Dequeued后,我们就应该立即释放锁了,如果我们在执行task的时候,一直持有锁,则会没有必要的阻塞其他线程来获取任务。


Wait Timeouts

在调用Wait方法的时候可以传递一个毫秒或Timespan的时间来设置超时。如果Wait超时了,那么Wait方法就会返回false。

带有超时功能的Wait方法的主要步骤:

    释放锁。
    阻塞 直到 pulsed 或者超时。
    重新获取锁。

超时就好像CLR 在超时到了的时候自动的调用了 pulse方法一样。

下面是使用超时的Wait的主要代码:

         lock(_locker)

         while()

                   Monitor.Wait(_locker,);


Monitor.Wait 方法返回一个bool值来代表是调用了pulse还是已经超时了。

如果是true: 代表调用了pulse。

如果是false:代表超时了。

这对记录日志很有用。


    
 
 
 
本站(WWW.)旨在分享和传播互联网科技相关的资讯和技术,将尽最大努力为读者提供更好的信息聚合和浏览方式。
本站(WWW.)站内文章除注明原创外,均为转载、整理或搜集自网络。欢迎任何形式的转载,转载请注明出处。












  • 相关文章推荐
  • Docker支持更深入的容器日志分析
  • 关于《深入浅出MFC》
  • Linux有没有什么好的高级的书,我要深入,
  • 深入理解linux内核
  • [100分]有没有关于binutils的深入的资料?或者深入底层的资料?
  • 深入理解PHP内核 TIPI
  • 操作系统 iis7站长之家
  • 哪位有《JSP深入编程》电子版?
  • 想要深入学习LINUX该学什么?
  • 100分求:哪儿有《深入理解linux内核》可供下哉!
  • 如何深入Linux的内核学习?
  • U-BOOT得掌握到什么程序,用不用深入去学
  • 想深入了解操作系统该怎么做
  • 前一阵子学习了shell脚本,如果想深入点了解linux可以看什么书呢
  • 问一个《深入理解计算机系统》中的问题
  • 深入多线程之:深入分析Interlocked
  • ##想买书深入学习linux下的编程,请指教
  • 深入JDBC sqlserver连接写法的详解
  • 深入oracle特定信息排序的分析
  • 深入分析C中不安全的sprintf与strcpy
  • 哪儿有下载《深入理解Linux内核》这本书?(中文)


  • 站内导航:


    特别声明:169IT网站部分信息来自互联网,如果侵犯您的权利,请及时告知,本站将立即删除!

    ©2012-2021,,E-mail:www_#163.com(请将#改为@)

    浙ICP备11055608号-3