快捷導(dǎo)航

深入多線程之:深入生產(chǎn)者、消費(fèi)者隊(duì)列分析

更新時間：2013年05月14日 15:13:06 作者：

本篇文章是對生產(chǎn)者與消費(fèi)者隊(duì)列進(jìn)行了詳細(xì)的分析介紹，需要的朋友參考下

上次我們使用AutoResetEvent實(shí)現(xiàn)了一個生產(chǎn)/消費(fèi)者隊(duì)列。這一次我們要使用Wait和Pulse方法來實(shí)現(xiàn)一個更強(qiáng)大的版本，它允許多個消費(fèi)者，每一個消費(fèi)者都在自己的線程中運(yùn)行。

我們使用數(shù)組來跟蹤線程。

Thread[] _workers;

通過跟蹤線程可以讓我們在所有的線程都結(jié)束后再結(jié)束我們的隊(duì)列任務(wù)。

每一個消費(fèi)者線程都執(zhí)行一個叫做Consume的方法，在一個for循環(huán)中，我們可以創(chuàng)建和啟動線程。例如：

復(fù)制代碼代碼如下:

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

上次我們使用的是一個字符串來代表任務(wù)，這次我們使用Action委托，它的定義如下：

Public delegate void Action();

為了表示一系列的任務(wù)，我們使用Queue<T> 集合，例如：

Queue<Action> _itemQ = new Queue<Action>();

在我們調(diào)用生產(chǎn)(EnqueueItem)和消費(fèi)(Consume)方法前，還是完整的看一看代碼吧：

復(fù)制代碼代碼如下:

class PCQueue
    {
        readonly object _locker = new object();
        Thread[] _workers;
        Queue<Action> _itemQ = new Queue<Action>(); //保存任務(wù)的隊(duì)列
        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); //通知等待隊(duì)列中的線程
            }
        }

        void Consume()
        {
            while (true)
            {
                Action item;
                lock (_locker)
                {
                    while (_itemQ.Count == 0)
                    {
                        Monitor.Wait(_locker); //釋放鎖，并阻止當(dāng)前線程，直到其他線程發(fā)送pulse信號。                    }
                    item = _itemQ.Dequeue();
                }

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

我們可以有一個退出策略，插入一個null item作為consumer退出的信號。如果我們想要快速的退出，可以使用一個獨(dú)立的”cancel” 標(biāo)記，因?yàn)槲覀冎С侄鄠€consumers，所以我們必須為每一個consumer插入一個null item。

下面是Main方法。使用兩個consumer線程，然后讓這兩個consumers執(zhí)行10個委托。

復(fù)制代碼代碼如下:

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); //等待關(guān)閉
            Console.WriteLine();
            Console.WriteLine("Workers complete!");
        }

下面讓我們細(xì)致的看一看EnqueueItem方法:

復(fù)制代碼代碼如下:

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

因?yàn)槲覀兊年?duì)列_itemQ被多線程環(huán)境使用，因此在對_itemQ進(jìn)行讀取的時候需要加鎖lock.

因?yàn)槲覀儾迦肓艘粋€新的任務(wù)，我們必須修改阻塞條件，也就是調(diào)用pulse方法，來喚醒調(diào)用了wait方法的線程。

出于對效率的考慮，當(dāng)插入一個Item的時候使用Pulse來代替PulseAll方法,因?yàn)榇蟛糠謺r候每一個Item只需要一個consumer來執(zhí)行。如果你有一個冰淇淋，你不可能叫30個睡眠的孩子都起來吃它，同樣，對于一個item，同時喚醒30個consumers一點(diǎn)好處都沒有。

讓我們再看看Consumer方法。

我們希望當(dāng)沒什么事情做的時候，線程阻塞就可以了，換句話說，隊(duì)列中沒有item的時候，線程就應(yīng)該阻塞。因此我們的阻塞條件是_itemQ.Count ==0;

復(fù)制代碼代碼如下:

Action item;
                lock (_locker)
                {
                    while (_itemQ.Count == 0)
                    {
                        Monitor.Wait(_locker); //釋放鎖，并阻止當(dāng)前線程，直到其他線程發(fā)送pulse信號。                    }
                    item = _itemQ.Dequeue();
                }

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

while循環(huán)退出的時候也意味著_itemQ 至少有一個item。我們必須在釋放鎖之前調(diào)用你哦個dequeue方法來獲取item，考慮下下面的代碼：

復(fù)制代碼代碼如下:

lock (_locker)
                {
                    while (_itemQ.Count == 0)
                    {
                        Monitor.Wait(_locker); //釋放鎖，并阻止當(dāng)前線程，直到其他線程發(fā)送pulse信號。                    }
                }
                //現(xiàn)在在這里可能被搶占，_itemQ可能被修改
                lock (_locker)
                {
                    item = _itemQ.Dequeue();
                }

在item被Dequeued后，我們就應(yīng)該立即釋放鎖了，如果我們在執(zhí)行task的時候，一直持有鎖，則會沒有必要的阻塞其他線程來獲取任務(wù)。

Wait Timeouts

在調(diào)用Wait方法的時候可以傳遞一個毫秒或Timespan的時間來設(shè)置超時。如果Wait超時了，那么Wait方法就會返回false。

帶有超時功能的Wait方法的主要步驟：

    釋放鎖。
    阻塞直到 pulsed 或者超時。
    重新獲取鎖。

超時就好像CLR 在超時到了的時候自動的調(diào)用了 pulse方法一樣。

下面是使用超時的Wait的主要代碼:

lock(_locker)

while(<阻塞條件>)

Monitor.Wait(_locker,<超時時間>);

Monitor.Wait 方法返回一個bool值來代表是調(diào)用了pulse還是已經(jīng)超時了。

如果是true：代表調(diào)用了pulse。