可重复使用地并行数据结构和算法

1、WORD格式9 种可重复使用的并行数据构造和算法目录倒计数锁存(Countdown Latch)可重用旋转等待(Spin Wait)屏障(Barrier)阻塞队列受限缓冲区(Bounded Buffer)Thin 事件无锁定LIFO堆栈循环分块并行分拆总结本专栏并未涉及很多公共语言运行库(CLR)功能的机制问题，而是更多介绍了如何有效使用您手头所具有的工具。 身为一名程序员，必须做出很多决策， 而选择正确的 数据构造和算法 无疑是最常见的， 也是最重要的决策之一。错误的选择可能导致程序无法运行，而大多数情况下， 那么决定了性能的好坏。 鉴于并行编程通常旨在改进性能，并且要难于串行编程，因此所作

2、的选择对您程序的成功就更为重要。在本专栏中， 我们将介绍九种可重复使用的数据构造和算法，这些构造和算法是许多并行程序所常用的，您应该能够轻松将它们应用到自己的.NET 软件中。专栏中每个例如随附的代码都是可用的，但尚未经过完全定型、测试和优化。这里列举的模式虽然并不详尽，但却代表了一些较为常见的模式。如您所见，很多例如都是互为补充的。在开场前，我想还是先介绍一些相关内容。Microsoft" .NET Framework提供了几个现有的并发基元。 虽然我要为您讲解如何构建自己的基元，但实际上现有基元是足以应付大多数情况的。 我只是想说某些可选的方案有时也是有参考价值的。此外，了解这些

3、技巧如何应用于实际操作也有助于加深您对并行编程的整体理解。在开场讲解前， 我假定您对现有基元已经有了一个根本的了解。您也可以参阅"MSDN"杂志" 2005年 8 月版的文章关于多线程应用程序：每个开发人员都应了解的内容，以全面了解其概念。一、倒计数锁存(Countdown Latch)专业资料整理WORD格式Semaphore 之所以成为并发编程中一种较为知名的数据构造，只是因为它在计算机科学领域有着悠久的历史可以追溯到19原因是多方面的， 而并不世纪 60 年代的操作系统设专业资料整理WORD格式计。Semaphore只是一种带有一个计数字段的数据构造，它只支

4、持两种操作：放置和取走专业资料整理WORD格式通常分别称为P 和V 。一次放置操作会增加一个semaphore计数，而一次取走操作专业资料整理WORD格式会减少一个 semaphore 计数。当 semaphore 计数为零时，除非执行一项 并发的放置操作使计数变为非零值 ，否那么任何后续的取走尝试都将阻塞等待。这两种操作均为不可再分(atomic)操作，并发时不会产生错误，能够确保并发的放置和取走操作有序地进展。Windows专业资料整理WORD格式具有根底内核和对semaphore对象的Win32支持请参阅CreateSemaphore和相关API ，专业资料整理WORD格式并且在.NET

5、 Framework中这些对象可以通过System.Threading.Semaphore类公开到上专业资料整理WORD格式层。 Mutex和Monitor所支持的临界区，通常被认为是一种特殊的semaphore，其计数会专业资料整理WORD格式在 0 和1 之间来回切换，换句话说，是一个二进制的semaphore。专业资料整理WORD格式另外还有一种反向semaphore也是非常有用。也就是说，有时您需要数据构造能够等待数据构造计数归零。Fork/join式并行模式在数据并行编程中是极为常见的，其中由单个主线程控制执行假设干辅助 线程并等待线程执行完毕。在这类情况下，使用反向semaphor

6、e 会很有帮助。大多数时候，您其实并不想唤醒线程来修改计数。因此在这种情况专业资料整理WORD格式下，我们将构造称为倒计数锁存 ，用来表示计数的减少，同时还说明一旦设置为Signaled专业资料整理WORD格式状态， 锁存将保持signaled这是一个与锁存相关的属性。遗憾的是，Windows和.NET专业资料整理WORD格式Framework 均不支持这种数据构造。但令人欣慰的是，构建这种数据闭锁并不难。专业资料整理WORD格式要构建倒计数锁存，只需将其计数器初始值设为n，并让每项辅助任务在完成时不可再专业资料整理WORD格式分地将n 减掉一个计数，这可以通过为减量操作加上锁或调用Inter

7、locked.Decrement来专业资料整理WORD格式实现。 接下来， 线程可以不执行取走操作，而是减少计数并等待计数器归零；而当线程被唤专业资料整理WORD格式醒时，它就可以得知已经有n 个信号向锁存注册。在while (count != 0)循环中，让等待的专业资料整理WORD格式线程阻塞通常是不错的选择这种情况下，您稍后将不得不使用事件，而不是使用旋转。public class CountdownLatch private int m_remain;private EventWaitHandle m_event;public CountdownLatch(int count) m_r

8、emain = count;m_event = new ManualResetEvent(false);public void Signal()/ The last thread to signal also sets the event. if (Interlocked.Decrement(ref m_remain) = 0)m_event.Set();专业资料整理WORD格式public void Wait() m_event.WaitOne();这看上去极为简单，但要正确运用还需要技巧。稍后我们将通过一些例如来讲解如何使用这种数据构造。请注意，此处所示根本实现还有很多可以改进地方，例如：

9、在事件上调用WaitOne 之前添加某种程度的旋转等待、缓慢分配事件而不是在构造器中进展分配以防足够的旋转会防止出现阻塞，如本专栏稍后介绍的ThinEvent演示的那样、添加重置功能以及提供Dispose 方法以便在不再需要内部事件对象时将对象关闭。二、可重用旋转等待(Spin Wait)虽然忙碌等待(busy waiting)更容易实现阻塞，但在某些情况下，您也许确实想在退回到真正的等待状态前先旋转(spin) 一段时间。我们很难理解为何这样做会有帮助，而大多数人之所以一开场就防止旋转等待，是因为旋转看上去像是在做无用功；如果上下文切换 每当线程等待内核事件时都会发生需要几千个周期在Wind

10、ows上确实是这样，我们称之为c，并且线程所等待的条件出现的时间少于2c 周期时间 1c 用于等待自身，1c 用于唤醒 ，那么旋转可以降低等待所造成的系统开销和滞后时间，从而提升算法的整体吞吐量和可伸缩性。如果您决定使用旋转等待，就必须慎重行事。因为如果这样做，您可能需要注意很多问题，比方：要确保在旋转循环内调用Thread.SpinWait ，以提高Intel 超线程技术的计算机上硬件对其他硬件线程的可用性；偶尔使用参数1 而非0 来调用Thread.Sleep，以防止优先级反向问题；通过轻微的回退(back-off)来引入随机选择，从而改善访问的局部性假定调用方持续重读共享状态并可能防止活

11、锁；当然， 在单CPU 的计算机最好不要采用这种方法因为在这种环境下旋转是非常浪费资源的。SpinWait类需要被定义为值类型，以便分配起来更加节省资源。现在，我们可以使用此算法来防止前述CountdownLatch算法中出现的阻塞。public struct SpinWait private int m_count;private static readonly bool s_isSingleProc =(Environment.ProcessorCount = 1);private const int s_yieldFrequency = 4000;private const int s_

12、yieldOneFrequency = 3*s_yieldFrequency;专业资料整理WORD格式public int Spin() 专业资料整理WORD格式int oldCount = m_count;/ On a single-CPU machine, we ensure our counter is always/ a multiple of "s_yieldFrequency , so we yield every time./ Else, we just increment by one.m_count += (s_isSingleProc " s_yield

13、Frequency : 1);/ If not a multiple of "s_yieldFrequency spin (w/ backoff).int countModFrequency = m_count % s_yieldFrequency;if (countModFrequency > 0)Thread.SpinWait(int)(1 + (countModFrequency * 0.05f);elseThread.Sleep(m_count <= s_yieldOneFrequency " 0 : 1);return oldCount;private

14、 void Yield()Thread.Sleep(m_count < s_yieldOneFrequency " 0 : 1);private const int s_spinCount = 4000;public void Wait()SpinWait s = new SpinWait();while (m_remain > 0)if (s.Spin() >= s_spinCount) m_event.WaitOne();专业资料整理WORD格式不可否认，选择频率和旋转计数是不确定的。与Win32临界区旋转计数类似，我们应专业资料整理WORD格式该根据测试和实验

15、的结果来选择合理的数值，而且即使合理的数值在不同系统中也会发生变专业资料整理WORD格式化。例如，根据Microsoft Media Center和Windows kernel团队的经历，MSDN文档建议专业资料整理WORD格式临界区旋转计数为4,000 ，但您的选择可以有所不同。理想的计数取决于多种因素，包括专业资料整理WORD格式在给定时间等待事件的线程数和事件出现的频率等。大多数情况下， 您会希望通过等待事件来消除显式让出时间，如锁存的例如中所示。您甚至可以选择动态调整计数：例如，从中等数量的旋转开场，每次旋转失败就增加计数。一旦计数到达预定的最大值，就完全停顿旋转并立即发出WaitOn

16、e 。逻辑如下所示：您希望立即增加到达预定的最大周期数，但却无法超过最大周期数。如果您发现此最大值缺乏以阻止上下文切换， 那么立即执行上下文切换总的算来占用的资源更少。慢慢您就会希望旋转计数能够到达一个稳定的值。三、屏障(Barrier)屏障， 又称集合点，是一种并发性基元，它无需另一 主 线程控制即可实现各线程之间专业资料整理WORD格式简单的互相协调。每个线程在到达屏障时都会不可再分地发出信号并等待。仅当所有n 都专业资料整理WORD格式到达屏障时，才允许所有线程继续。这种方法可用于协调算法(cooperative algorithms)，该专业资料整理WORD格式算法广泛应用于科学、数学

17、和图形领域。 很多计算中都适合使用屏障，实际上，甚至CLR 的垃圾收集器都在使用它们。屏障只是将较大的计算分割为假设干较小的协作阶段(cooperativephase)，例如：const int P = .;Barrier barrier = new Barrier(P);Data partitions = new DataP;/ Runn ing on "P separate threads in parallel: public void Body(int myIndex) FillMyPartition(partitionsmyIndex);barrier.Await();Re

18、adOtherPartition(partitionsP myIndex - 1); barrier.Await();/ .您会很快发现在这种情况下是可以使用倒计数锁存的。每个线程都可以在调用Signal后立即调用Wait ，而不是调用Await ；在到达屏障后，所有线程都会被释放。但这里存在专业资料整理WORD格式一个问题： 前面所讲的锁存并不支持屡次重复使用同一对象，而这却是所有屏障都支持的一专业资料整理WORD格式个常用属性。 实际上， 上述例如就需要使用此属性。您可以通过单独的屏障对象来实现这一专业资料整理WORD格式点，但这样做非常浪费资源；而由于所有线程每次只出现在一个阶段，因此根

19、本无需多个屏专业资料整理WORD格式障对象。专业资料整理WORD格式要解决这个问题， 您可以使用一样的根底倒计数锁存算法来处理减少计数器计数、发信号指示事件、等待等方面的问题，从而将其扩展为可重复使用。要实现这一点，您需要使用所谓的感应反向屏障(sense reversing barrier)，该屏障需要在偶数 和 奇数 阶段之间交替。 在交替阶段需要使用单独的事件。using System;using System.Threading;public class Barrier private volatile int m_count;private int m_originalCount;p

20、rivate EventWaitHandle m_oddEvent;private EventWaitHandle m_evenEvent;private volatile bool m_sense = false; / false=even, true=odd.public Barrier(int count) m_count = count;m_originalCount = count;m_oddEvent = new ManualResetEvent(false);m_evenEvent = new ManualResetEvent(false);public void Await()

21、 bool sense = m_sense;/ The last thread to signal also sets the event.if (m_count = 1 | Interlocked.Decrement(ref m_count) = 0) m_count = m_originalCount;m_sense = !sense; / Reverse the sense.if (sense = true) / oddm_evenEvent.Reset();m_oddEvent.Set(); else / evenm_oddEvent.Reset();m_evenEvent.Set()

22、;专业资料整理WORD格式专业资料整理WORD格式 else if (sense = true) m_oddEvent.WaitOne();elsem_evenEvent.WaitOne();为何需要两个事件，其原因很难讲清楚。一种方法是在Await中先执行Set 随后立即执行Reset，但这很危险，并会导致死锁。原因有二：第一，另一线程的m_count 可能已减少，但在依次快速调用Set 和 Reset 时，计数的值还缺乏以到达可在事件上调用WaitOne 。第二，虽然等待的线程可能已经能够调用WaitOne ，但可报警等待如CLR一贯使用的那些 可以中断等待， 从而暂时从等待队列中删除等待

23、的线程，以便运行异步过程调用(APC) 。等待的线程将始终看不到事件处于设置(set) 状态。两种情况均会导致事件丢失，并可能导致死锁。针对奇数阶段和偶数阶段使用单独的事件即可防止这种情况。您可能想继续像CountdownLatch中那样将旋转添加到Barrier 。但如果您尝试这样做，可能会遇到一个问题：一般情况下，旋转线程会开场旋转直到m_count 归零。但通过上述实现， m_count 的值实际上永远不会为零，除非最后一个线程将其重置为m_originalCount 。这种想当然的方法将导致一个或多个线程进展旋转永远地 ，而其他线程那么会在下一阶段阻塞也是永远地。解决方法很简单。您旋转

24、，等待感应发生变化public void Await() bool sense = m_sense;/ The last thread to signal also sets the event.if (m_count = 1 | Interlocked.Decrement(ref m_count) = 0) m_count = m_originalCount;m_sense = !sense; / Reverse the sense.if (sense = true) / oddm_evenEvent.Set();m_oddEvent.Reset(); else / evenm_oddEve

25、nt.Set();m_evenEvent.Reset(); else 专业资料整理WORD格式SpinWait s = new SpinWait();while (sense = m_sense) if (s.Spin() >= s_spinCount) if (sense = true) m_oddEvent.WaitOne();elsem_evenEvent.WaitOne();由于所有线程都必须离开前一阶段的 Await 才可以完成下一阶段，因此可以确定，所有线程要么会观察到感应发生变化，要么最终等待事件并被唤醒。阻塞队列在共享内存的体系构造中，两项或多项任务间唯一的同步点通常是一

26、个位于中枢的共享集合的数据构造。通常，一项或多项任务会负责为其他一项或多项任务生成要执行的工作 ，我们称之为生产者 /使用者关系 (producer/consumer) 。这类数据构造的简单同步通常是非常简单的使用 Monitor 或 ReaderWriterLock 即可解决，但当缓冲区为空时，任务间的协调就会变得比较困难。要解决这个问题，通常需要使用阻塞队列。实际上，阻塞队列有几种稍微不同的变体，包括简单变体和复杂变体。在简单变量中， 使用者仅在队列为空时才会阻塞，而在复杂变量中，每个生产者都正好配有 一个使用者， 也就是说， 在使用者到达并开场处理排队工程之前，生产者会被阻塞，同理，在生

27、产者交付工程前，所有使用者也会被阻塞。这时通常使用FIFO 先进先出排序，但我们并不总是有必要这么做。 我们也可以对缓冲区进展限制，这一点稍后会为大家介绍。 由于稍后将要介绍的受限缓冲区包含更为简单的为空时阻塞 (blocking-when-empty)行为，因此我们这里仅对配对变量做一了解。要实现这个目的，我们只需封装一个简单的Queue<T> ，上方保持同步。那么到底需要何种同步？每当线程对元素进展排队时，在返回前都会等待使用者取消元素排队。当线程取消元素排队时，如果发现缓冲区为空，那么必须等待新元素的进入。当然，取消排队后，使用者必须通知生产者已取到其工程请参见图 5。Fig

28、ure 5 阻塞队列复制代码class Cell<T> internal T m_obj;internal Cell(T obj) m_obj = obj; 专业资料整理WORD格式public class BlockingQueue<T> private Queue<Cell<T>> m_queue = new Queue<Cell<T>>(); public void Enqueue(T obj) Cell<T> c = new Cell<T>(obj);lock (m_queue) m_que

29、ue.Enqueue(c);Monitor.Pulse(m_queue);Monitor.Wait(m_queue);public T Dequeue() Cell<T> c;lock (m_queue) while (m_queue.Count = 0)Monitor.Wait(m_queue);c = m_queue.Dequeue();Monitor.Pulse(m_queue);return c.m_obj;专业资料整理WORD格式请注意，我们可以在Enqueue方法中依次调用Pulse和Wait ，类似地，在Dequeue方法专业资料整理WORD格式中依次调用Wait和

30、Pulse。只有在释放监视器之后，才会对内部事件进展设置，而由于监专业资料整理WORD格式视器的这种实现方法，会导致某个线程调度ping-pong 。我们可能会想，也许可以使用Win32专业资料整理WORD格式事件来构建一个更加优化的通知机制。但是，使用这类完善的Win32事件会带来巨大开销，专业资料整理WORD格式尤其是使用它们时所进展的本钱分配和内核跳转(kernel transitions)，因此还是考虑其他选专业资料整理WORD格式择吧。您可以像CLR 的 ReaderWriterLock那样将这些时间集中到池中，也可以像ThinEvent类型稍后介绍一样缓慢分配它们。这种实现方法也是

31、有弊端的，即要为每个新元素分配对象，备选方法可能也会将这些对象参加到池中，但同样会带来其他麻烦。受限缓冲区(Bounded Buffer)某些类别的队列中会出现资源使用问题。 如果生产者任务创立工程的速度快于使用者处理工程的速度，那么系统的内存使用将不受限制。专业资料整理WORD格式为了说明这个问题， 我们假设一个系统，单个生产者每秒钟可将50个工程排入队列， 而使用者每秒钟只能使用10 个工程。 首先， 这样会打乱系统的平衡性，而且对于一对一的生产者 使用者配置无法进展很好的调整。只需一分钟，就会有2,400 个工程堆积在缓冲区中。假设这些工程中每个工程使用10KB内存，那将意味着缓冲区本身

32、就会占用24MB 内存。一小时后， 这个数字将飙升至1GB以上。解决这一问题的一个方法是调整生产者线程与使用者线程的比例，在上述情况中， 要将比例调整为一个生产者对应五个使用者。但是到达速度通常是不稳定的， 这会导致系统周期性的不稳定，并带来一些明显的问题，简单的固定比例是无法解决这个问题的。效劳器上的程序通常是长时间运行的，人们希望程序能够长期保持良好的运行状态，但如果内存使用不受限，就会造成不可防止的混乱，还可能导致必须定期回收效劳器进程的情况。受限缓冲区允许您对生产者强制阻塞前缓冲区可能到达的大小进展限制。阻塞生产者可让使用者有时机 赶上 通过允许其线程接收调度时间片，同时还能够解决内存

33、使用量增加的问题。我们的方法还是仅封装Queue<T>，同时添加两个等待条件和两个事件通知条件：生产者在队列满时等待，直至队列变为非满，而使用者在队列空时等待，直至变为非空； 生产者在生产出工程后通知等待的使用者，而使用者也会在取走工程后通知生产者，如图 6 中所示。Figure 6 受限缓冲区(Bounded Buffer)复制代码public class BoundedBuffer<T> private Queue<T> m_queue = new Queue<T>();private int m_consumersWaiting;priva

34、te int m_producersWaiting;private const int s_maxBufferSize = 128;public void Enqueue(T obj) lock (m_queue) while (m_queue.Count = (s_maxBufferSize - 1) m_producersWaiting+;Monitor.Wait(m_queue);m_producersWaiting-;m_queue.Enqueue(obj);if (m_consumersWaiting > 0)Monitor.PulseAll(m_queue);专业资料整理WO

35、RD格式public T Dequeue() T e;lock (m_queue) while (m_queue.Count = 0) m_consumersWaiting+;Monitor.Wait(m_queue);m_consumersWaiting-;e = m_queue.Dequeue();if (m_producersWaiting > 0)Monitor.PulseAll(m_queue);return e;这里我们再一次使用了一种比较天真的方法。但是我们确实优化了对PulseAll的调用因为它们消耗的资源很多，方法是使用m_consumersWaiting 和 m_pr

36、oducersWaiting这两个计数器并仅就计数器值是否为零发出信号。除此以外， 我们还可以再做进一步的改进。例如，共享与此类似的单个事件可能会唤醒过多线程：如果使用者将队列大小降为0，并且生产者和使用者同时处于等待状态，显然您只希望唤醒生产者至少开场时要这么做。此实现将以先进先出的规那么处理所有等待者，这意味着您可能需要在任一生产者运行前唤醒使用者，这样做仅仅是为了让使用者发现队列为空，然后再次等待。令人欣慰的是，最终出现生产者和使用者同时等待的情况是很少见的，但其出现也是有规律的，而且受限区域较小。Thin 事件与 Monitor.Wait 、 Pulse 和 PulseAll相比， W

37、in32事件有一个很大的优势：它们是粘滞的(sticky)。也就是说，一旦有事件收到信号，任何后续等待都将立即取消阻止，即使线程在信号发出前尚未开场等待。如果没有这个功能，您就需要经常编写一些代码将所有等待和信号通知严格地限制在临界区域内，由于Windows方案程序总是会提升已唤醒线程的优先级，因此这些代码的效率很低；如果不采取这种方法，您就必须使用某种技巧型很强、容易出现竞态条件的代码。专业资料整理WORD格式作为替代方法， 您可以使用thin 事件，这是一种可重用数据构造，可在阻塞前短暂地旋转，专业资料整理WORD格式仅在必要时才缓慢分配Win32事件，否那么允许您执行类似手动重置事件的行

38、为。换句话说，专业资料整理WORD格式它可以对那些复杂的容易导致竞态条件的代码进展封装，这样您就不必在整个根本代码内散专业资料整理WORD格式播它。此例如依赖于Vance Morrison的文章中描述的一些内存模型保证，使用的时候要多专业资料整理WORD格式加小心请参见图7。专业资料整理WORD格式Figure 7 Thin事件复制代码public struct ThinEvent private int m_state; / 0 means unset, 1 means set.private EventWaitHandle m_eventObj;private const int s_sp

39、inCount = 4000;public void Set() m_state = 1;Thread.MemoryBarrier(); / required.if (m_eventObj != null)m_eventObj.Set();public void Reset() m_state = 0;if (m_eventObj != null)m_eventObj.Reset();public void Wait() SpinWait s = new SpinWait();while (m_state = 0) if (s.Spin() >= s_spinCount) if (m_e

40、ventObj = null) ManualResetEvent newEvent =new ManualResetEvent(m_state = 1);if (InterlockedpareExchange<EventWaitHandle>(ref m_eventObj, newEvent, null) = null) 专业资料整理WORD格式/ If someone set the flag before seeing the new/ event obj, we must ensure it s been set.if (m_state = 1)m_eventObj.Set(

41、); else / Lost the race w/ another thread. Just use/ its event.newEvent.Close();m_eventObj.WaitOne();这根本上反映了 m_state 变量中的事件状态，其中值为0 表示未设置，值为1 表示已设置。现在，等待一个已设置事件消耗资源是很低的；如果m_state 在 Wait例程的入口处值为 1，我们会立即返回， 无需任何内核跳转。 但如果线程在事件设置完毕之前进入等待状态，处理上就需要很多技巧。 要等待的首个线程必须分配一个新的事件对象，并对其进展比较后交换至m_eventObj 字段中； 如果 C

42、AS 失败，那么意味着其他等待者对事件进展了初始化，所以我们只可重复使用它；否那么就必须重新检查自上次看到m_state 后其值是否发生更改。不然的话， m_state 的值也许会为1，那么 m_eventObj 就无法收到信号通知，这将导致在调用WaitOne 时出现死锁。调用Set 的线程必须首先设置m_state，随后如果发现值为非空的m_eventObj ，就会调用其上的 Set。这里需要两个内存屏障：需要注意的是切不可提前移动m_state 的第二次读取，我们会使用InterlockedpareExchange 设置m_eventObj来保证这点；在写入m_eventObj 之前，不

43、可移动Set 中的对 m_eventObj的读取这是一种在某些Intel 和 AMD 处理器以及CLR 2.0内存模型上出现的奇怪的合法转换，并未显式调用Thread.MemoryBarrier 。并行重置事件通常是不平安的，因此调用方需要进展额外的同步化。现在您可以轻松在其他地方使用它，例如在上述的CountdownLatch例如和队列例如中，而且通常这样做可以大幅度提升性能，尤其是当您可以巧妙地运用旋转时。我们上面介绍了一个技巧性很强的实现方式。请注意， 您可以使用单个标志和监视器来实现自动和手动重置类型，这远比本例如简单但效率方面有时会不及本例。专业资料整理WORD格式无锁定LIFO堆栈

44、专业资料整理WORD格式使用锁定构建线程平安的集合是相当直接明了的，即使限制和阻塞方面会有些复杂如上面看到的。但是，当所有协调均通过简单的后进先出(LIFO)堆栈数据构造实现时，使用锁定的开销会比实际所需的高。线程的临界区即保持锁定的时间有开场点和完毕点， 其持续时间可能会跨越许多指令的有效期。保持锁定可阻止其他线程同时进展读取和写入操作。这样做可以实现序列化， 这当然是我们所想要的结果，但严格地讲， 这种序列化要比我们所需的序列化强很多。 我们的目的是要在堆栈上推入和弹出元素，而这两项操作只需通过常规读取和单一的比较交换写入即可实现。我们可以利用这点来构建伸缩性更强的无锁定堆栈，从而不会强制

45、线程进展没必要的等待。我们的算法工作方式如下。 使用有的列表代表堆栈，其标头代表堆栈的顶端，并存储在m_head 字段中。在将一个新项推入堆栈时，要构造一个新节点，其值等于您要推入堆栈的值，然后从本地读取m_head 字段，并将其存储至新节点的m_next 字段，随后执行不可再分的 InterlockedpareExchange 来替换堆栈当前的头部。如果顶端自首次读取后在序列中的任何点发生更改， CompareExchange都会失败，并且线程必须返回循环并再次尝试整个序列。弹出同样是比较直截了当的。读取m_head 并尝试将其与 m_next 引用的本地副本交换；如果失败，您只需一直尝试，

46、如图 8 中所示。Win32提供了一种名为SList 的类比数据构造，其构建方法采用了类似的算法。Figure 8 无锁定堆栈复制代码public class LockFreeStack<T> private volatile StackNode<T> m_head;public void Push(T item) StackNode<T> node = new StackNode<T>(item);StackNode<T> head;do head = m_head;node.m_next = head; while (m_head

47、 != head | InterlockedpareExchange( ref m_head, node, head) != head);public T Pop() StackNode<T> head;SpinWait s = new SpinWait();专业资料整理WORD格式while (true) StackNode<T> next;do head = m_head;if (head = null) goto emptySpin;next = head.m_next; while (m_head != head | InterlockedpareExchang

48、e( ref m_head, next, head) != head);break;emptySpin:s.Spin();return head.m_value;class StackNode<T> internal T m_value;internal StackNode<T> m_next;internal StackNode(T val) m_value = val; 请注意， 这是理想的并发控制的一种形式：无需阻止其他线程存取数据，只需抱着会在争用中胜出 的信念继续即可。如果事实证明无法胜出 ，您将会遇到一些变化不定的问题，例如活锁。这种设计方案还说明您无法确保能够实现FIFO 调度。根据概率统计，系统中所有线程都将向前推进。 而实际上从整体来看，系统进度总是向前推进的，因为其中一个线程的失败总是意味着至少有一个其他线程是推进的这是调用该无锁定 的一个条件