可重復(fù)使用的并行數(shù)據(jù)結(jié)構(gòu)和算法

1、9種可重復(fù)使用的并行數(shù)據(jù)結(jié)構(gòu)和算法目錄倒計(jì)數(shù)鎖存(CountdownLatch)可重用旋轉(zhuǎn)等待(SpinWait)屏障(Barrier)阻塞隊(duì)列受限緩沖區(qū)(BoundedBuffer)Thin事件無(wú)鎖定LIFO堆棧循環(huán)分塊并行分拆總結(jié)本專(zhuān)欄并未涉及很多公共語(yǔ)言運(yùn)行庫(kù)(CLR)功能的機(jī)制問(wèn)題，而是更多介紹了如何有效使用您手頭所具有的工具。身為一名程序員，必須做出很多決策，而選擇正確的數(shù)據(jù)結(jié)構(gòu)和算法無(wú)疑是最常見(jiàn)的，也是最重要的決策之一。錯(cuò)誤的選擇可能導(dǎo)致程序無(wú)法運(yùn)行，而大多數(shù)情況下，則決定了性能的好壞。鑒于并行編程通常旨在改進(jìn)性能，并且要難于串行編程，因此所作的選擇對(duì)您程序的成功就更為重要。在本專(zhuān)

2、欄中，我們將介紹九種可重復(fù)使用的數(shù)據(jù)結(jié)構(gòu)和算法，這些結(jié)構(gòu)和算法是許多并行程序所常用的，您應(yīng)該能夠輕松將它們應(yīng)用到自己的.NET軟件中。專(zhuān)欄中每個(gè)示例隨附的代碼都是可用的，但尚未經(jīng)過(guò)完全定型、測(cè)試和優(yōu)化。這里列舉的模式雖然并不詳盡，但卻代表了一些較為常見(jiàn)的模式。如您所見(jiàn)，很多示例都是互為補(bǔ)充的。在開(kāi)始前，我想還是先介紹一些相關(guān)內(nèi)容。Microsoft?.NETFramework提供了幾個(gè)現(xiàn)有的并發(fā)基元。雖然我要為您講解如何構(gòu)建自己的基元，但實(shí)際上現(xiàn)有基元是足以應(yīng)付大多數(shù)情況的。我只是想說(shuō)某些可選的方案有時(shí)也是有參考價(jià)值的。此外，了解這些技巧如何應(yīng)用于實(shí)際操作也有助于加深您對(duì)并行編程的整體理解。在

3、開(kāi)始講解前，我假定您對(duì)現(xiàn)有基元已經(jīng)有了一個(gè)基本的了解。您也可以參閱MSDN?雜志2005年8月版的文章關(guān)于多線(xiàn)程應(yīng)用程序：每個(gè)開(kāi)發(fā)人員都應(yīng)了解的內(nèi)容”，以全面了解其概念。、倒計(jì)數(shù)鎖存(CountdownLatch)Semaphore之所以成為并發(fā)編程中一種較為知名的數(shù)據(jù)結(jié)構(gòu)，原因是多方面的，而并不只是因?yàn)樗谟?jì)算機(jī)科學(xué)領(lǐng)域有著悠久的歷史(可以追溯到19世紀(jì)60年代的操作系統(tǒng)設(shè)計(jì))。Semaphore只是一種帶有一個(gè)計(jì)數(shù)字段的數(shù)據(jù)結(jié)構(gòu)，它只支持兩種操作：放置和取走(通常分別稱(chēng)為P和V)。一次放置操作會(huì)增加一個(gè)semaphore計(jì)數(shù)，而一次取走操作會(huì)減少一個(gè)semaphore計(jì)數(shù)。當(dāng)semapho

4、re計(jì)數(shù)為零時(shí)，除非執(zhí)行一項(xiàng)并發(fā)的放置操作使計(jì)數(shù)變?yōu)榉橇阒?，否則任何后續(xù)的取走嘗試都將阻塞(等待)。這兩種操作均為不可再分(atomic)操作，并發(fā)時(shí)不會(huì)產(chǎn)生錯(cuò)誤，能夠確保并發(fā)的放置和取走操作有序地進(jìn)行。Windows具有基礎(chǔ)內(nèi)核和對(duì)semaphore對(duì)象的Win32支持(請(qǐng)參閱CreateSemaphore和相關(guān)API),并且在.NETFramework中這些對(duì)象可以通過(guò)System.Threading.Semaphore類(lèi)公開(kāi)到上層。Mutex和Monitor所支持的臨界區(qū)，通常被認(rèn)為是一種特殊的semaphore,其計(jì)數(shù)會(huì)在0和1之間來(lái)回切換，換句話(huà)說(shuō)，是一個(gè)二進(jìn)制的semaphore。

5、另外還有一種反向semaphore”也是非常有用。也就是說(shuō)，有時(shí)您需要數(shù)據(jù)結(jié)構(gòu)能夠等待數(shù)據(jù)結(jié)構(gòu)計(jì)數(shù)歸零。Fork/join式并行模式在數(shù)據(jù)并行編程中是極為常見(jiàn)的，其中由單個(gè)主”線(xiàn)程控制執(zhí)行若干輔助”線(xiàn)程并等待線(xiàn)程執(zhí)行完畢。在這類(lèi)情況下，使用反向semaphore會(huì)很有幫助。大多數(shù)時(shí)候，您其實(shí)并不想喚醒線(xiàn)程來(lái)修改計(jì)數(shù)。因此在這種情況下，我們將結(jié)構(gòu)稱(chēng)為倒計(jì)數(shù)鎖存”，用來(lái)表示計(jì)數(shù)的減少，同時(shí)還表明一旦設(shè)置為“Signaled狀態(tài)，鎖存將保持“signaled(這是一個(gè)與鎖存相關(guān)的屬性)。遺憾的是，Windows和.NETFramework均不支持這種數(shù)據(jù)結(jié)構(gòu)。但令人欣慰的是，構(gòu)建這種數(shù)據(jù)閉鎖并不難。

6、要構(gòu)建倒計(jì)數(shù)鎖存，只需將其計(jì)數(shù)器初始值設(shè)為n,并讓每項(xiàng)輔助任務(wù)在完成時(shí)不可再分地將n減掉一個(gè)計(jì)數(shù)，這可以通過(guò)為減量操作加上鎖”或調(diào)用Interlocked.Decrement來(lái)實(shí)現(xiàn)。接下來(lái)，線(xiàn)程可以不執(zhí)行取走操作，而是減少計(jì)數(shù)并等待計(jì)數(shù)器歸零；而當(dāng)線(xiàn)程被喚醒時(shí)，它就可以得知已經(jīng)有n個(gè)信號(hào)向鎖存注冊(cè)。在while(count!=0)循環(huán)中，讓等待的線(xiàn)程阻塞通常是不錯(cuò)的選擇(這種情況下，您稍后將不得不使用事件)，而不是使用旋轉(zhuǎn)。publicclassCountdownLatchprivateintm_remain;privateEventWaitHandlem_event;publicCountd

7、ownLatch(intcount)m_remain=count;m_event=newManualResetEvent(false);publicvoidSignal()/Thelastthreadtosignalalsosetstheevent.if(Interlocked.Decrement(refm_remain)=0)m_event.Set();publicvoidWait（）m_event.WaitOne（）;這看上去極為簡(jiǎn)單，但要正確運(yùn)用還需要技巧。稍后我們將通過(guò)一些示例來(lái)講解如何使用這種數(shù)據(jù)結(jié)構(gòu)。請(qǐng)注意，此處所示基本實(shí)現(xiàn)還有很多可以改進(jìn)地方，例如：在事件上調(diào)用WaitOne之前

8、添加某種程度的旋轉(zhuǎn)等待、緩慢分配事件而不是在構(gòu)造器中進(jìn)行分配（以防足夠的旋轉(zhuǎn)會(huì)避免出現(xiàn)阻塞，如本專(zhuān)欄稍后介紹的ThinEvent演示的那樣）、添加重置功能以及提供Dispose方法（以便在不再需要內(nèi)部事件對(duì)象時(shí)將對(duì)象關(guān)閉）。二、可重用旋轉(zhuǎn)等待（SpinWait）雖然忙碌等待（busywaiting）更容易實(shí)現(xiàn)阻塞，但在某些情況下，您也許的確想在退回到真正的等待狀態(tài)前先旋轉(zhuǎn)（spin）一段時(shí)間。我們很難理解為何這樣做會(huì)有幫助，而大多數(shù)人之所以一開(kāi)始就避免旋轉(zhuǎn)等待，是因?yàn)樾D(zhuǎn)看上去像是在做無(wú)用功；如果上下文切換（每當(dāng)線(xiàn)程等待內(nèi)核事件時(shí)都會(huì)發(fā)生）需要幾千個(gè)周期（在Windows上確實(shí)是這樣），我們稱(chēng)

9、之為c,并且線(xiàn)程所等待的條件出現(xiàn)的時(shí)間少于2c周期時(shí)間（1c用于等待自身，1c用于喚醒），則旋轉(zhuǎn)可以降低等待所造成的系統(tǒng)開(kāi)銷(xiāo)和滯后時(shí)間，從而提升算法的整體吞吐量和可伸縮性。如果您決定使用旋轉(zhuǎn)等待，就必須謹(jǐn)慎行事。因?yàn)槿绻@樣做，您可能需要注意很多問(wèn)題，比如：要確保在旋轉(zhuǎn)循環(huán)內(nèi)調(diào)用Thread.SpinWait,以提高Intel超線(xiàn)程技術(shù)的計(jì)算機(jī)上硬件對(duì)其他硬件線(xiàn)程的可用性；偶爾使用參數(shù)1而非0來(lái)調(diào)用Thread.Sleep,以避免優(yōu)先級(jí)反向問(wèn)題；通過(guò)輕微的回退（back-off）來(lái)引入隨機(jī)選擇，從而改善訪(fǎng)問(wèn)的局部性（假定調(diào)用方持續(xù)重讀共享狀態(tài)）并可能避免活鎖；當(dāng)然，在單CPU的計(jì)算機(jī)最好不要采

10、用這種方法（因?yàn)樵谶@種環(huán)境下旋轉(zhuǎn)是非常浪費(fèi)資源的）。SpinWait類(lèi)需要被定義為值類(lèi)型，以便分配起來(lái)更加節(jié)省資源?，F(xiàn)在，我們可以使用此算法來(lái)避免前述CountdownLatch算法中出現(xiàn)的阻塞。publicstructSpinWaitprivateintm_count;privatestaticreadonlybools_isSingleProc=（Environment.ProcessorCount=1）;privateconstints_yieldFrequency=4000;privateconstints_yieldOneFrequency=3*s_yieldFrequency;pu

11、blicintSpin（）intoldCount=m_count;/Onasingle-CPUmachine,weensureourcounterisalways/amultipleof's_yieldFrequency',soweyieldeverytime./Else,wejustincrementbyone.m_count+=(s_isSingleProc?s_yieldFrequency:1);/Ifnotamultipleof's_yieldFrequency'spin(w/backoff).intcountModFrequency=m_count%s

12、_yieldFrequency;if(countModFrequency>0)Thread.SpinWait(int)(1+(countModFrequency*0.05f);elseThread.Sleep(m_count<=s_yieldOneFrequency?0:1);returnoldCount;privatevoidYield()Thread.Sleep(m_count<s_yieldOneFrequency?0:1);privateconstints_spinCount=4000;publicvoidWait()SpinWaits=newSpinWait();w

13、hile(m_remain>0)if(s.Spin()>=s_spinCount)m_event.WaitOne();不可否認(rèn)，選擇頻率和旋轉(zhuǎn)計(jì)數(shù)是不確定的。與Win32臨界區(qū)旋轉(zhuǎn)計(jì)數(shù)類(lèi)似，我們應(yīng)該根據(jù)測(cè)試和實(shí)驗(yàn)的結(jié)果來(lái)選擇合理的數(shù)值，而且即使合理的數(shù)值在不同系統(tǒng)中也會(huì)發(fā)生變化。例如，根據(jù)MicrosoftMediaCenter和Windowskernel團(tuán)隊(duì)的經(jīng)驗(yàn)，MSDN文檔建議臨界區(qū)旋轉(zhuǎn)計(jì)數(shù)為4,000，但您的選擇可以有所不同。理想的計(jì)數(shù)取決于多種因素，包括在給定時(shí)間等待事件的線(xiàn)程數(shù)和事件出現(xiàn)的頻率等。大多數(shù)情況下，您會(huì)希望通過(guò)等待事件來(lái)消除顯式讓出時(shí)間，如鎖存的示例中所示。

14、您甚至可以選擇動(dòng)態(tài)調(diào)整計(jì)數(shù)：例如，從中等數(shù)量的旋轉(zhuǎn)開(kāi)始，每次旋轉(zhuǎn)失敗就增加計(jì)數(shù)。一旦計(jì)數(shù)達(dá)到預(yù)定的最大值，就完全停止旋轉(zhuǎn)并立即發(fā)出WaitOne。邏輯如下所示：您希望立即增加達(dá)到預(yù)定的最大周期數(shù)，但卻無(wú)法超過(guò)最大周期數(shù)。如果您發(fā)現(xiàn)此最大值不足以阻止上下文切換，那么立即執(zhí)行上下文切換總的算來(lái)占用的資源更少。慢慢您就會(huì)希望旋轉(zhuǎn)計(jì)數(shù)能夠達(dá)到一個(gè)穩(wěn)定的值。屏障（Barrier）屏障，又稱(chēng)集合點(diǎn)，是一種并發(fā)性基元，它無(wú)需另一生”線(xiàn)程控制即可實(shí)現(xiàn)各線(xiàn)程之間簡(jiǎn)單的互相協(xié)調(diào)。每個(gè)線(xiàn)程在到達(dá)屏障時(shí)都會(huì)不可再分地發(fā)出信號(hào)并等待。僅當(dāng)所有n都到達(dá)屏障時(shí)，才允許所有線(xiàn)程繼續(xù)。這種方法可用于協(xié)調(diào)算法(cooperati

15、vealgorithms),該算法廣泛應(yīng)用于科學(xué)、數(shù)學(xué)和圖形領(lǐng)域。很多計(jì)算中都適合使用屏障，實(shí)際上，甚至CLR的垃圾收集器都在使用它們。屏障只是將較大的計(jì)算分割為若干較小的協(xié)作階段(cooperativephase),例如：constintP=；Barrierbarrier=newBarrier(P);Data口partitions=newDataP;/Runningon'P'separatethreadsinparallel:publicvoidBody(intmyIndex)FillMyPartition(partitionsmyIndex);barrier.Await()

16、;ReadOtherPartition(partitionsP-myIndex-1);barrier.Await();/.您會(huì)很快發(fā)現(xiàn)在這種情況下是可以使用倒計(jì)數(shù)鎖存的。每個(gè)線(xiàn)程都可以在調(diào)用Signal后立即調(diào)用Wait,而不是調(diào)用Await;在到達(dá)屏障后，所有線(xiàn)程都會(huì)被釋放。但這里存在一個(gè)問(wèn)題：前面所講的鎖存并不支持多次重復(fù)使用同一對(duì)象，而這卻是所有屏障都支持的一個(gè)常用屬性。實(shí)際上，上述示例就需要使用此屬性。您可以通過(guò)單獨(dú)的屏障對(duì)象來(lái)實(shí)現(xiàn)這一點(diǎn)，但這樣做非常浪費(fèi)資源；而由于所有線(xiàn)程每次只出現(xiàn)在一個(gè)階段，因此根本無(wú)需多個(gè)屏障對(duì)象。要解決這個(gè)問(wèn)題，您可以使用相同的基礎(chǔ)倒計(jì)數(shù)鎖存算法來(lái)處理減少計(jì)數(shù)

17、器計(jì)數(shù)、發(fā)信號(hào)指示事件、等待等方面的問(wèn)題，從而將其擴(kuò)展為可重復(fù)使用。要實(shí)現(xiàn)這一點(diǎn)，您需要使用所謂的感應(yīng)反向屏障(sensereversingbarrier),該屏障需要在偶數(shù)”和奇數(shù)”階段之間交替。在交替階段需要使用單獨(dú)的事件。usingSystem;usingSystem.Threading;publicclassBarrierprivatevolatileintm_count;privateintm_originalCount;privateEventWaitHandlem_oddEvent;privateEventWaitHandlem_evenEvent;privatevolatile

18、boolm_sense=false;/false=even,true=odd.publicBarrier(intcount)m_count=count;m_originalCount=count;m_oddEvent=newManualResetEvent(false);m_evenEvent=newManualResetEvent(false);publicvoidAwait()boolsense=m_sense;/Thelastthreadtosignalalsosetstheevent.if(m_count=1|Interlocked.Decrement(refm_count)=0)m_

19、count=m_originalCount;m_sense=!sense;Reversethesense.if(sense=true)/oddm_evenEvent.Reset();m_oddEvent.Set();else/evenm_oddEvent.Reset();m_evenEvent.Set();elseif（sense=true）m_oddEvent.WaitOne（）;elsem_evenEvent.WaitOne（）;為何需要兩個(gè)事件，其原因很難講清楚。一種方法是在A(yíng)wait中先執(zhí)行Set隨后立即執(zhí)行Reset,但這很危險(xiǎn)，并會(huì)導(dǎo)致死鎖。原因有二：第一，另一線(xiàn)程的m_count

20、可能已減少，但在依次快速調(diào)用Set和Reset時(shí)，計(jì)數(shù)的值還不足以達(dá)到可在事件上調(diào)用WaitOne。第二，雖然等待的線(xiàn)程可能已經(jīng)能夠調(diào)用WaitOne,但可報(bào)警等待（如CLR-貫使用的那些）可以中斷等待，從而暫時(shí)從等待隊(duì)列中刪除等待的線(xiàn)程，以便運(yùn)行異步過(guò)程調(diào)用（APC）。等待的線(xiàn)程將始終看不到事件處于設(shè)置（set）狀態(tài)。兩種情況均會(huì)導(dǎo)致事件丟失，并可能導(dǎo)致死鎖。針對(duì)奇數(shù)階段和偶數(shù)階段使用單獨(dú)的事件即可避免這種情況。您可能想繼續(xù)像CountdownLatch中那樣將旋轉(zhuǎn)添加到Barrier。但如果您嘗試這樣做，可能會(huì)遇到一個(gè)問(wèn)題：一般情況下，旋轉(zhuǎn)線(xiàn)程會(huì)開(kāi)始旋轉(zhuǎn)直到m_count歸零。但通過(guò)上述實(shí)

21、現(xiàn)，m_count的值實(shí)際上永遠(yuǎn)不會(huì)為零，除非最后一個(gè)線(xiàn)程將其重置為m_originalCount。這種想當(dāng)然的方法將導(dǎo)致一個(gè)或多個(gè)線(xiàn)程進(jìn)行旋轉(zhuǎn)（永遠(yuǎn)地），而其他線(xiàn)程則會(huì)在下一階段阻塞（也是永遠(yuǎn)地）。解決方法很簡(jiǎn)單。您旋轉(zhuǎn)，等待感應(yīng)發(fā)生變化publicvoidAwait()boolsense=m_sense;/Thelastthreadtosignalalsosetstheevent.if(m_count=1|Interlocked.Decrement(refm_count)=0)m_count=m_originalCount;m_sense=!sense;/Reversethesense.i

22、f(sense=true)/oddm_evenEvent.Set();m_oddEvent.Reset();else/evenm_oddEvent.Set();m_evenEvent.Reset();elseSpinWaits=newSpinWait();while(sense=m_sense)if(s.Spin()>=s_spinCount)if(sense=true)m_oddEvent.WaitOne();elsem_evenEvent.WaitOne();由于所有線(xiàn)程都必須離開(kāi)前一階段的Await才可以完成下一階段，因此可以確定，所有線(xiàn)程要么會(huì)觀(guān)察到感應(yīng)發(fā)生變化，要么最終等待事件

23、并被喚醒。阻塞隊(duì)列在共享內(nèi)存的體系結(jié)構(gòu)中，兩項(xiàng)或多項(xiàng)任務(wù)間唯一的同步點(diǎn)通常是一個(gè)位于中樞的共享集合的數(shù)據(jù)結(jié)構(gòu)。通常，一項(xiàng)或多項(xiàng)任務(wù)會(huì)負(fù)責(zé)為其他一項(xiàng)或多項(xiàng)任務(wù)生成要執(zhí)行的工作”，我們稱(chēng)之為生產(chǎn)者/使用者關(guān)系(producer/consumer)。這類(lèi)數(shù)據(jù)結(jié)構(gòu)的簡(jiǎn)單同步通常是非常簡(jiǎn)單的一使用Monitor或ReaderWriterLock即可解決，但當(dāng)緩沖區(qū)為空時(shí)，任務(wù)間的協(xié)調(diào)就會(huì)變得比較困難。要解決這個(gè)問(wèn)題，通常需要使用阻塞隊(duì)列。實(shí)際上，阻塞隊(duì)列有幾種稍微不同的變體，包括簡(jiǎn)單變體和復(fù)雜變體。在簡(jiǎn)單變量中，使用者僅在隊(duì)列為空時(shí)才會(huì)阻塞，而在復(fù)雜變量中，每個(gè)生產(chǎn)者都正好配有”一個(gè)使用者，也就是說(shuō)，在

24、使用者到達(dá)并開(kāi)始處理排隊(duì)項(xiàng)目之前，生產(chǎn)者會(huì)被阻塞，同理，在生產(chǎn)者交付項(xiàng)目前，所有使用者也會(huì)被阻塞。這時(shí)通常使用FIFO(先進(jìn)先出)排序，但我們并不總是有必要這么做。我們也可以對(duì)緩沖區(qū)進(jìn)行限制，這一點(diǎn)稍后會(huì)為大家介紹。由于稍后將要介紹的受限緩沖區(qū)包含更為簡(jiǎn)單的為空時(shí)阻塞"(blockingwhen-empty)行為，因此我們這里僅對(duì)配對(duì)變量做一了解。要實(shí)現(xiàn)這個(gè)目的，我們只需封裝一個(gè)簡(jiǎn)單的Queue<T>,上方保持同步。那么到底需要何種同步？每當(dāng)線(xiàn)程對(duì)元素進(jìn)行排隊(duì)時(shí)，在返回前都會(huì)等待使用者取消元素排隊(duì)。當(dāng)線(xiàn)程取消元素排隊(duì)時(shí)，如果發(fā)現(xiàn)緩沖區(qū)為空，則必須等待新元素的進(jìn)入。當(dāng)然，取

25、消排隊(duì)后，使用者必須通知生產(chǎn)者已取到其項(xiàng)目(請(qǐng)參見(jiàn)圖5)。Figure5阻塞隊(duì)列復(fù)制代碼classCell<T>internalTm_obj;internalCell(Tobj)m_obj=obj;publicclassBlockingQueue<T>privateQueue<Cell<T>>m_queue=newQueue<Cell<T>>();publicvoidEnqueue"obj)Cell<T>c=newCell<T>(obj);lock(m_queue)m_queue.Enqu

26、eue(c);Monitor.Pulse(m_queue);Monitor.Wait(m_queue);publicTDequeue。Cell<T>c;lock(m_queue)while(m_queue.Count=0)Monitor.Wait(m_queue);c=m_queue.Dequeue();Monitor.Pulse(m_queue);returnc.m_obj;請(qǐng)注意，我們可以在Enqueue方法中依次調(diào)用Pulse和Wait,類(lèi)似地，在Dequeue方法中依次調(diào)用Wait和Pulseo只有在釋放監(jiān)視器之后，才會(huì)對(duì)內(nèi)部事件進(jìn)行設(shè)置，而由于監(jiān)視器的這種實(shí)現(xiàn)方法，會(huì)導(dǎo)致

27、某個(gè)線(xiàn)程調(diào)度ping-pong。我們可能會(huì)想，也許可以使用Win32事件來(lái)構(gòu)建一個(gè)更加優(yōu)化的通知機(jī)制。但是，使用這類(lèi)完善的Win32事件會(huì)帶來(lái)巨大開(kāi)銷(xiāo)，尤其是使用它們時(shí)所進(jìn)行的成本分配和內(nèi)核跳轉(zhuǎn)(kerneltransitions),因此還是考慮其他選擇吧。您可以像CLR的ReaderWriterLock那樣將這些時(shí)間集中到池中，也可以像ThinEvent類(lèi)型(稍后介紹)一樣緩慢分配它們。這種實(shí)現(xiàn)方法也是有弊端的，即要為每個(gè)新元素分配對(duì)象，備選方法可能也會(huì)將這些對(duì)象加入到池中，但同樣會(huì)帶來(lái)其他麻煩。受限緩沖區(qū)(BoundedBuffer)某些類(lèi)別的隊(duì)列中會(huì)出現(xiàn)資源使用問(wèn)題。如果生產(chǎn)者任務(wù)創(chuàng)建項(xiàng)

28、目的速度快于使用者處理項(xiàng)目的速度，則系統(tǒng)的內(nèi)存使用將不受限制。為了說(shuō)明這個(gè)問(wèn)題，我們假設(shè)一個(gè)系統(tǒng)，單個(gè)生產(chǎn)者每秒鐘可將50個(gè)項(xiàng)目排入隊(duì)列，而使用者每秒鐘只能使用10個(gè)項(xiàng)目。首先，這樣會(huì)打亂系統(tǒng)的平衡性，而且對(duì)于一對(duì)一的生產(chǎn)者一使用者配置無(wú)法進(jìn)行很好的調(diào)整。只需一分鐘，就會(huì)有2,400個(gè)項(xiàng)目堆積在緩沖區(qū)中。假設(shè)這些項(xiàng)目中每個(gè)項(xiàng)目使用10KB內(nèi)存，那將意味著緩沖區(qū)本身就會(huì)占用24MB內(nèi)存。一小時(shí)后，這個(gè)數(shù)字將飆升至1GB以上。解決這一問(wèn)題的一個(gè)方法是調(diào)整生產(chǎn)者線(xiàn)程與使用者線(xiàn)程的比例，在上述情況中，要將比例調(diào)整為一個(gè)生產(chǎn)者對(duì)應(yīng)五個(gè)使用者。但是到達(dá)速度通常是不穩(wěn)定的，這會(huì)導(dǎo)致系統(tǒng)周期性的不穩(wěn)定，并帶來(lái)

29、一些明顯的問(wèn)題，簡(jiǎn)單的固定比例是無(wú)法解決這個(gè)問(wèn)題的。服務(wù)器上的程序通常是長(zhǎng)時(shí)間運(yùn)行的，人們希望程序能夠長(zhǎng)期保持良好的運(yùn)行狀態(tài)，但如果內(nèi)存使用不受限，就會(huì)造成不可避免的混亂，還可能導(dǎo)致必須定期回收服務(wù)器進(jìn)程的情況。受限緩沖區(qū)允許您對(duì)生產(chǎn)者強(qiáng)制阻塞前緩沖區(qū)可能達(dá)到的大小進(jìn)行限制。阻塞生產(chǎn)者可讓使用者有機(jī)會(huì)趕上”(通過(guò)允許其線(xiàn)程接收調(diào)度時(shí)間片)，同時(shí)還能夠解決內(nèi)存使用量增加的問(wèn)題。我們的方法還是僅封裝Queue<T>,同時(shí)添加兩個(gè)等待條件和兩個(gè)事件通知條件：生產(chǎn)者在隊(duì)列滿(mǎn)時(shí)等待，直至隊(duì)列變?yōu)榉菨M(mǎn)，而使用者在隊(duì)列空時(shí)等待，直至變?yōu)榉强眨簧a(chǎn)者在生產(chǎn)出項(xiàng)目后通知等待的使用者，而使用者也會(huì)在取

30、走項(xiàng)目后通知生產(chǎn)者，如圖6中所示。Figure 6 受限緩沖區(qū)(BoundedBuffer)復(fù)制代碼publicclassBoundedBuffer<T>privateQueue<T>m_queue=newQueue<T>();privateintm_consumersWaiting;privateintm_producersWaiting;privateconstints_maxBufferSize=128;publicvoidEnqueue"obj)lock(m_queue)while(m_queue.Count=(s_maxBufferSiz

31、e-1)m_producersWaiting+;Monitor.Wait(m_queue);m_producersWaiting-;m_queue.Enqueue(obj);if(m_consumersWaiting>0)Monitor.PulseAll(m_queue);publicTDequeue。Te；lock(m_queue)while(m_queue.Count=0)m_consumersWaiting+;Monitor.Wait(m_queue);m_consumersWaiting-;e=m_queue.Dequeue();if(m_producersWaiting>

32、0)Monitor.PulseAll(m_queue);returne;這里我們?cè)僖淮问褂昧艘环N比較天真的方法。但是我們確實(shí)優(yōu)化了對(duì)PulseAll的調(diào)用(因?yàn)樗鼈兒馁M(fèi)的資源很多)，方法是使用m_consumersWaiting和m_producersWaiting這兩個(gè)計(jì)數(shù)器并僅就計(jì)數(shù)器值是否為零發(fā)出信號(hào)。除此以外，我們還可以再做進(jìn)一步的改進(jìn)。例如，共享與此類(lèi)似的單個(gè)事件可能會(huì)喚醒過(guò)多線(xiàn)程：如果使用者將隊(duì)列大小降為0,并且生產(chǎn)者和使用者同時(shí)處于等待狀態(tài)，顯然您只希望喚醒生產(chǎn)者(至少開(kāi)始時(shí)要這么做)。此實(shí)現(xiàn)將以先進(jìn)先出的規(guī)則處理所有等待者，這意味著您可能需要在任一生產(chǎn)者運(yùn)行前喚醒使用者，這樣做

33、僅僅是為了讓使用者發(fā)現(xiàn)隊(duì)列為空，然后再次等待。令人欣慰的是，最終出現(xiàn)生產(chǎn)者和使用者同時(shí)等待的情況是很少見(jiàn)的，但其出現(xiàn)也是有規(guī)律的，而且受限區(qū)域較小。Thin事件與Monitor.Wait、Pulse和PulseAll相比，Win32事件有一個(gè)很大的優(yōu)勢(shì)：它們是粘滯的"(sticky)也就是說(shuō)，一旦有事件收到信號(hào)，任何后續(xù)等待都將立即取消阻止，即使線(xiàn)程在信號(hào)發(fā)出前尚未開(kāi)始等待。如果沒(méi)有這個(gè)功能，您就需要經(jīng)常編寫(xiě)一些代碼將所有等待和信號(hào)通知嚴(yán)格地限制在臨界區(qū)域內(nèi)，由于Windows計(jì)劃程序總是會(huì)提升已喚醒線(xiàn)程的優(yōu)先級(jí)，因此這些代碼的效率很低；如果不采取這種方法，您就必須使用某種技巧型很強(qiáng)

34、、容易出現(xiàn)競(jìng)態(tài)條件的代碼。作為替代方法，您可以使用“thin事件”，這是一種可重用數(shù)據(jù)結(jié)構(gòu)，可在阻塞前短暫地旋轉(zhuǎn)，僅在必要時(shí)才緩慢分配Win32事件，否則允許您執(zhí)行類(lèi)似手動(dòng)重置事件的行為。換句話(huà)說(shuō),它可以對(duì)那些復(fù)雜的容易導(dǎo)致競(jìng)態(tài)條件的代碼進(jìn)行封裝，這樣您就不必在整個(gè)基本代碼內(nèi)散播它。此示例依賴(lài)于VanceMorrison的文章中描述的一些內(nèi)存模型保證，使用的時(shí)候要多加小心（請(qǐng)參見(jiàn)圖7）。Figure 7 Thin事件復(fù)制代碼publicstructThinEventprivateintm_state;0meansunset,1meansset.privateEventWaitHandlem_e

35、ventObj;privateconstints_spinCount=4000;publicvoidSet()m_state=1;Thread.MemoryBarrier();/required.if(m_eventObj!=null)m_eventObj.Set();publicvoidReset()m_state=0;if(m_eventObj!=null)m_eventObj.Reset();publicvoidWait()SpinWaits=newSpinWait();while(m_state=0)if(s.Spin()>=s_spinCount)if(m_eventObj=n

36、ull)ManualResetEventnewEvent=newManualResetEvent(m_state=1);if(Interlocked.CompareExchange<EventWaitHandle(refm_eventObj,newEvent,null)=null)/Ifsomeonesettheflagbeforeseeingthenew/eventobj,wemustensureit'sbeenset.if(m_state=1)m_eventObj.Set();else/Losttheracew/anotherthread.Justuse/itsevent.n

37、ewEvent.Close();m_eventObj.WaitOne();這基本上反映了m_state變量中的事件狀態(tài)，其中值為0表示未設(shè)置，值為1表示已設(shè)置?，F(xiàn)在，等待一個(gè)已設(shè)置事件耗費(fèi)資源是很低的；如果m_state在Wait例程的入口處值為1,我們會(huì)立即返回，無(wú)需任何內(nèi)核跳轉(zhuǎn)。但如果線(xiàn)程在事件設(shè)置完畢之前進(jìn)入等待狀態(tài)，處理上就需要很多技巧。要等待的首個(gè)線(xiàn)程必須分配一個(gè)新的事件對(duì)象，并對(duì)其進(jìn)行比較后交換至m_eventObj字段中；如果CAS失敗，則意味著其他等待者對(duì)事件進(jìn)行了初始化，所以我們只可重復(fù)使用它；否則就必須重新檢查自上次看到m_state后其值是否發(fā)生更改。不然的話(huà)，m_sta

38、te的值也許會(huì)為1,那么m_eventObj就無(wú)法收到信號(hào)通知，這將導(dǎo)致在調(diào)用WaitOne時(shí)出現(xiàn)死鎖。調(diào)用Set的線(xiàn)程必須首先設(shè)置m_state,隨后如果發(fā)現(xiàn)值為非空的m_eventObj,就會(huì)調(diào)用其上的Set。這里需要兩個(gè)內(nèi)存屏障：需要注意的是切不可提前移動(dòng)m_state的第二次讀取，我們會(huì)使用Interlocked.CompareExchange設(shè)置m_eventObj來(lái)保證這點(diǎn)；在寫(xiě)入m_eventObj之前，不可移動(dòng)Set中的對(duì)m_eventObj的讀取(這是一種在某些Intel和AMD處理器以及CLR2.0內(nèi)存模型上出現(xiàn)的奇怪的合法轉(zhuǎn)換，并未顯式調(diào)用Thread.MemoryBar

39、rier)。并行重置事件通常是不安全的，因此調(diào)用方需要進(jìn)行額外的同步化?，F(xiàn)在您可以輕松在其他地方使用它，例如在上述的CountdownLatch示例和隊(duì)列示例中，而且通常這樣做可以大幅度提升性能，尤其是當(dāng)您可以巧妙地運(yùn)用旋轉(zhuǎn)時(shí)。我們上面介紹了一個(gè)技巧性很強(qiáng)的實(shí)現(xiàn)方式。請(qǐng)注意，您可以使用單個(gè)標(biāo)志和監(jiān)視器來(lái)實(shí)現(xiàn)自動(dòng)和手動(dòng)重置類(lèi)型，這遠(yuǎn)比本示例簡(jiǎn)單(但效率方面有時(shí)會(huì)不及本例)。無(wú)鎖定LIFO堆棧使用鎖定構(gòu)建線(xiàn)程安全的集合是相當(dāng)直接明了的，即使限制和阻塞方面會(huì)有些復(fù)雜(如上面看到的)。但是，當(dāng)所有協(xié)調(diào)均通過(guò)簡(jiǎn)單的后進(jìn)先出(LIFO)堆棧數(shù)據(jù)結(jié)構(gòu)實(shí)現(xiàn)時(shí)，使用鎖定的開(kāi)銷(xiāo)會(huì)比實(shí)際所需的高。線(xiàn)程的臨界區(qū)(即保

40、持鎖定的時(shí)間)有開(kāi)始點(diǎn)和結(jié)束點(diǎn)，其持續(xù)時(shí)間可能會(huì)跨越許多指令的有效期。保持鎖定可阻止其他線(xiàn)程同時(shí)進(jìn)行讀取和寫(xiě)入操作。這樣做可以實(shí)現(xiàn)序列化，這當(dāng)然是我們所想要的結(jié)果，但嚴(yán)格地講，這種序列化要比我們所需的序列化強(qiáng)很多。我們的目的是要在堆棧上推入和彈出元素，而這兩項(xiàng)操作只需通過(guò)常規(guī)讀取和單一的比較一交換寫(xiě)入即可實(shí)現(xiàn)。我們可以利用這點(diǎn)來(lái)構(gòu)建伸縮性更強(qiáng)的無(wú)鎖定堆棧，從而不會(huì)強(qiáng)制線(xiàn)程進(jìn)行沒(méi)必要的等待。我們的算法工作方式如下。使用有鏈接的列表代表堆棧，其標(biāo)頭代表堆棧的頂端，并存儲(chǔ)在m_head字段中。在將一個(gè)新項(xiàng)推入堆棧時(shí)，要構(gòu)造一個(gè)新節(jié)點(diǎn)，其值等于您要推入堆棧的值，然后從本地讀取m_head字段，并將其存

41、儲(chǔ)至新節(jié)點(diǎn)的m_next字段，隨后執(zhí)行不可再分的Interlocked.CompareExchange來(lái)替換堆棧當(dāng)前的頭部。如果頂端自首次讀取后在序列中的任何點(diǎn)發(fā)生更改，CompareExchange都會(huì)失敗，并且線(xiàn)程必須返回循環(huán)并再次嘗試整個(gè)序列。彈出同樣是比較直截了當(dāng)?shù)?。讀取m_head并嘗試將其與m_next引用的本地副本交換；如果失敗，您只需一直嘗試，如圖8中所示。Win32提供了一種名為SList的類(lèi)比數(shù)據(jù)結(jié)構(gòu)，其構(gòu)建方法采用了類(lèi)似的算法。Figure8無(wú)鎖定堆棧復(fù)制代碼publicclassLockFreeStack<T>privatevolatileStackNode

42、<T>m_head;publicvoidPush(Titem)StackNode<T>node=newStackNode<T>(item);StackNode<T>head;dohead=m_head;node.m_next=head;while(m_head!=head|Interlocked.CompareExchange(refm_head,node,head)!=head);publicTPop()StackNode<T>head;SpinWaits=newSpinWait();while(true)StackNode<

43、T>next;dohead=m_head;if(head=null)gotoemptySpin;next=head.m_next;while(m_head!=head|Interlocked.CompareExchange(refm_head,next,head)!=head);break;emptySpin:s.Spin();returnhead.m_value;classStackNode<T>internalTm_value;internalStackNode<T>m_next;internalStackNode(Tval)m_value=val;請(qǐng)注意，這

44、是理想的并發(fā)控制的一種形式：無(wú)需阻止其他線(xiàn)程存取數(shù)據(jù)，只需抱著會(huì)在爭(zhēng)用中勝出”的信念繼續(xù)即可。如果事實(shí)證明無(wú)法勝出”，您將會(huì)遇到一些變化不定的問(wèn)題，例如活鎖。這種設(shè)計(jì)方案還表明您無(wú)法確保能夠?qū)崿F(xiàn)FIFO調(diào)度。根據(jù)概率統(tǒng)計(jì)，系統(tǒng)中所有線(xiàn)程都將向前推進(jìn)。而實(shí)際上從整體來(lái)看，系統(tǒng)進(jìn)度總是向前推進(jìn)的，因?yàn)槠渲幸粋€(gè)線(xiàn)程的失敗總是意味著至少有一個(gè)其他線(xiàn)程是推進(jìn)的(這是調(diào)用該先鎖定”的一個(gè)條件)。有些情況下，當(dāng)CompareExchange無(wú)法避免m_head大量爭(zhēng)用內(nèi)存時(shí)，使用指數(shù)回退算法會(huì)很有用。同時(shí)，我們還對(duì)堆棧變空的情況采取了相當(dāng)直截了當(dāng)?shù)姆椒?。我們只是一直旋轉(zhuǎn)，等待有新項(xiàng)目推入。將Pop重新寫(xiě)入處

45、于非等待狀態(tài)的TryPop是很簡(jiǎn)單易懂的，但是要利用事件進(jìn)行等待則會(huì)有一些復(fù)雜。這兩個(gè)功能都是十分重要的，所以留給那些喜歡動(dòng)手的讀者作為練習(xí)之用。我們?yōu)槊總€(gè)Push都分配了對(duì)象，這讓我們無(wú)需擔(dān)心所謂的ABA問(wèn)題。在內(nèi)部重復(fù)使用已從列表中彈出的節(jié)點(diǎn)就會(huì)導(dǎo)致ABA問(wèn)題。開(kāi)發(fā)人員有時(shí)會(huì)嘗試將節(jié)點(diǎn)集中到池中以減少對(duì)象分配的數(shù)目，但這樣做是有問(wèn)題的：結(jié)果會(huì)是，即使對(duì)m_head執(zhí)行了大量干擾性寫(xiě)入操作，一個(gè)不可再分割的操作也可以實(shí)現(xiàn)，但實(shí)際上卻是不正確的。（例如：線(xiàn)程1讀取節(jié)點(diǎn)A,然后由線(xiàn)程2將節(jié)點(diǎn)A刪除并置入池中；線(xiàn)程2將節(jié)點(diǎn)B作為新的頭推入，然后節(jié)點(diǎn)A從池中返回到線(xiàn)程2并被推入；隨后，線(xiàn)程1會(huì)成功執(zhí)

46、行CompareExchange,但現(xiàn)在作為頭的A已經(jīng)與先前所讀取的不同了。）嘗試以本機(jī)C/C+編寫(xiě)此算法時(shí)也會(huì)出現(xiàn)類(lèi)似問(wèn)題；因?yàn)橐坏┑刂繁会尫牛瑑?nèi)存分配器就會(huì)立即重復(fù)使用這些地址，節(jié)點(diǎn)會(huì)被彈出并釋放，然后該節(jié)點(diǎn)的地址會(huì)被用于新的節(jié)點(diǎn)分配，結(jié)果導(dǎo)致與上面相同的問(wèn)題。接下來(lái)我們不再討論ABA,有關(guān)它的詳細(xì)介紹我們可以從其他的資源獲得。最后，可以使用類(lèi)似的無(wú)鎖定技術(shù)編寫(xiě)一個(gè)FIFO隊(duì)列。它的優(yōu)勢(shì)在于并行推入和彈出的線(xiàn)程之間并不一定發(fā)生沖突，而在上述LockFreeStack中，推入者和彈出者始終會(huì)爭(zhēng)用同一m_head字段。然而，這種算法相當(dāng)復(fù)雜，如果您好奇的話(huà)，我推薦您閱讀MagedM.Micha

47、el和MichaelL.Scott在1996年撰寫(xiě)的文章Simple,Fast,andPracticalNon-BlockingandBlockingConcurrentQueueAlgorithms”。循環(huán)分塊循環(huán)分塊是指對(duì)循環(huán)的輸入范圍或數(shù)據(jù)進(jìn)行分區(qū)，并將每個(gè)分區(qū)分配給單獨(dú)的線(xiàn)程以實(shí)現(xiàn)并發(fā)。這是某些編程模型（例如OpenMP）實(shí)現(xiàn)并行性的一項(xiàng)基本技術(shù)（請(qǐng)參閱KangSuGatlin的MSDN雜志文章），通常稱(chēng)為并行Forall循環(huán)，是從高性能的FORTRAN術(shù)語(yǔ)中得到啟發(fā)而創(chuàng)造的。無(wú)論如何，范圍都只是一組索引：復(fù)制代碼foreach（Teinlist）我們都可以設(shè)計(jì)分區(qū)技術(shù)以提供分塊的循環(huán)

48、。您可能想應(yīng)用多種特定于數(shù)據(jù)結(jié)構(gòu)的分區(qū)技術(shù)，這些技術(shù)確實(shí)很多，本專(zhuān)欄無(wú)法一一列舉。因此這里我們只關(guān)注一種常用技術(shù)，可用于將數(shù)組中各種非連續(xù)范圍的元素分配到每個(gè)分區(qū)。我們只需計(jì)算跨距的值（它大約等于元素?cái)?shù)目除以分區(qū)數(shù)目的值），并用它來(lái)計(jì)算連續(xù)范圍（參見(jiàn)圖9）。當(dāng)然盡管其他方法是有效、有用并在某些情況下有必需的，但當(dāng)輸入內(nèi)容為值類(lèi)型數(shù)組時(shí)，此方法可以實(shí)現(xiàn)較好的空間局部性。Figure9循環(huán)分塊復(fù)制代碼publicstaticvoidForAll(intfrom,intto,Action<int>a,intp)ForAll<int>(null,from,to,null,a,p

49、);publicstaticvoidForAll<T>(IList<T>data,Action<T>a,intp)ForAll<T>(data,0,data.Count,a,null,p);privatestaticvoidForAll<T>(IList<T>data,intfrom,intto,Action<T>a0,Action<int>a1,intp)intsize=from-to;intstride=(size+p-1)/p;CountdownLatchlatch=newCountdownL

50、atch(p);for(inti=0;i<p;i+)intidx=i;ThreadPool.QueueUserWorkItem(delegateintend=Math.Min(size,stride*(idx+1);for(intj=stride*idx;j<end;j+)if(data!=null)a0(dataj);elsea1(j);latch.Signal(););latch.Wait();我們這里提供了兩種公共版的ForAll：一種接受一定范圍內(nèi)的數(shù)字，另一種接受IList<T>(類(lèi)似于C#中的Foreach循環(huán))。兩者都轉(zhuǎn)發(fā)到同一幫助程序重載，重載可調(diào)用從給

51、定索引的列表傳遞元素這一操作，或者傳遞索引本身。您應(yīng)使用第一個(gè)重載（通常在此加入一個(gè)普通For循環(huán)）。例如，下面的代碼復(fù)制代碼for（inti=0;i<10;i+）S;將變?yōu)椋簭?fù)制代碼Parallel.ForAll（0,10,delegate（inti）S;,Environment.ProcessorCount）;現(xiàn)在可以使用第二個(gè)重載（通常在此加入一個(gè)C#foreach循環(huán)），這樣，復(fù)制代碼List<T>list=；foreach（Teinlist）S;將變?yōu)椋簭?fù)制代碼Parallel.ForAll（list,delegate"e）S;,Environment.P

52、rocessorCount）;您需要謹(jǐn)慎以確保S中的語(yǔ)句不會(huì)寫(xiě)入共享內(nèi)存；否則您需要向并行版本添加合適的同步操作。當(dāng)然可以編寫(xiě)相應(yīng)版本來(lái)支持IEnumerable<T>、以其他方式對(duì)迭代空間進(jìn)行分區(qū)等（為了節(jié)省空間，本專(zhuān)欄省略了這些內(nèi)容）。在本示例中，在n項(xiàng)子任務(wù)的持續(xù)時(shí)間內(nèi)調(diào)用線(xiàn)程被浪費(fèi)”了。更好的方法是使用調(diào)用線(xiàn)程來(lái)運(yùn)行其中一項(xiàng)任務(wù)，然后在其完成時(shí)加入其他任務(wù)。沒(méi)有必要通過(guò)擴(kuò)展ForAll方法來(lái)執(zhí)行此操作。并行分拆有一類(lèi)操作可使用分拆（又稱(chēng)為折疊或聚合）來(lái)執(zhí)行，其中會(huì)以某種方式將許多值進(jìn)行組合以便生成單一輸出。一般分拆的工作方式如下。使用一個(gè)二元運(yùn)算符（即包含兩個(gè)參數(shù)的函數(shù)），

53、并對(duì)照向量或一組大小為n的元素從左至右對(duì)其進(jìn)行計(jì)算。對(duì)于j=0至n-1,調(diào)用該二元運(yùn)算符，作為輸入傳遞至jth迭代，并將在元素j-1上調(diào)用運(yùn)算符得到的輸出作為第一個(gè)參數(shù)，將jth元素本身作為第二個(gè)參數(shù)。由于沒(méi)有預(yù)先給定的值可用，因此會(huì)將一個(gè)特殊的種子值用作第0個(gè)元素的第一個(gè)參數(shù)。然后使用最終（可選）結(jié)果選擇器將中間值轉(zhuǎn)換為最終結(jié)果。我們來(lái)看一個(gè)示例。如果二元運(yùn)算符為+,輸入為包含5個(gè)元素1,2,3,4,5的向量，那么展開(kāi)的計(jì)算應(yīng)類(lèi)似于(1+2)+3)+4)+5)。如果將此展開(kāi)式轉(zhuǎn)換為函數(shù)調(diào)用格式，則類(lèi)似于以下形式(假定種子值為0):+(+(+(+(+(0,1),2),3),4),5)換句話(huà)說(shuō)，

54、您只需計(jì)算所有輸入數(shù)字的總和即可。這稱(chēng)為總和分拆。有種直接的方法可以將這種一般化的算法轉(zhuǎn)換為串行算法，如下所示：復(fù)制代碼delegateTFunc<T>(Targ0,Targl);TReduce<T>(Tinput,Tseed,Func<T>r)Tresult=seed;foreach(Teininput)result=r(result,e);returnresult;調(diào)用它時(shí)，您只需求得一組數(shù)字的和即可，如下所示(在C#3.0中)：復(fù)制代碼intnums=.somesetofnumbers；intsum=Reduce(nums,0,(x,y)=>x+y;);所有