.net 中的生产者消费者链和线程调度答案

【问题标题】：Producer Consumer Chain and Thread Scheduling in .net.net 中的生产者消费者链和线程调度
【发布时间】：2011-04-01 21:45:15
【问题描述】：

我创建了一个由 4 个生产者-消费者线程组成的链（形成一个 4 步管道）。令我惊讶的是，所有四个线程都按顺序运行，而不是同时运行！！！也就是说，第二个线程神秘地等待，直到第一个线程完全完成生产。第三个线程神秘地等待，直到第二个线程完全完成生产（依此类推）。

情况变得更糟。如果我将 Thread.Sleep(300) 放入第一个生产者的循环中，那么其他三个线程将变为并发并实际获得处理器时间，正如预期的那样，从多线程应用程序中产生“随机交错”控制台输出。我几乎无法接受“睡眠”是解决方案的必要部分的想法，但我看到在code written by Jon Skeet 中以这种方式完全结合了睡眠。

请告诉我实现并发不是必需的，或者如果是，那么为什么？

关于我的特定生产者-消费者链的更准确的故事如下所示：

第一个线程：在一个紧密的循环中，它尽可能快地生成“小部件”消息，并将它们推入队列以供下一个线程使用。当第一个小部件添加到队列中时，System.Threading.Timer 设置为 ~100 毫秒。计时器触发的回调是第二个线程...
第二个线程（从计时器触发）：从前一个队列中读取部分或全部小部件。它将它们发送到另一个队列（由第三个线程使用）。 monitor.Pulse/Wait 机制用于与第三个线程同步。
第三线程： 在监视器上阻塞。等到 monitor.Pulse 被调用，然后从队列中获取一项。一项被推送到最终队列中，推送完成后再次使用 monitor.Pulse。
第四个线程： 在监视器上阻塞。等到 monitor.Pulse 被调用。小部件已处理。

通过此管道处理 100 万个小部件大约需要 4 分钟。在 4 分钟内，有足够的时间安排最后 3 个线程并与第一个线程同时执行一些工作。但正如我所说，最后三个线程按顺序运行，除非我为第一个线程引入一个小睡眠。没有意义。

对为什么会这样工作有什么想法吗？

附言请不要告诉我，正如我所描述的那样，一条长的生产者-消费者链可以缩小或消除。请相信我（或假设）我需要这么长的链条。 :)

【问题讨论】：

看起来您遇到了锁定问题。如果没有详细的代码示例，不要指望任何更有用的答案。
最好使用提供的系统机制；特别是，BlockingCollection<T> 用于队列，Task 用于任务。
@Stephen：我使用的是 .net 3.5，它没有 BlockingCollection。
基于 Thread.Sleep() 的 XML 文档，显然这种“线程按顺序运行问题”是正常行为。 sleep 的文档说使用 '0' 毫秒“表示应该暂停该线程以允许其他等待线程执行。”奇怪的是，Thread.Sleep(1) 完全解决了这个问题，而 Thread.Sleep(0) 什么也没做。尽管如此，如果某些事情需要它（即我的问题），文档不会提供这意味着。不过，我认为现在这整个事情都是正常的行为——尽管它仍然让我觉得很疯狂。
@Mystagogue：Rx library 包括 .NET 3.5 的 BlockingCollection<T> 和 Task，如果您有兴趣的话。

标签： .net multithreading synchronization

【解决方案1】：

让我猜猜 - 您很可能使用某种锁定机制？我建议对消息队列进行无锁实现以提高您的性能。要么这样，要么你将它们全部强制到同一个硬件线程上，但我不确定你是否可以在 Windows 上做到这一点。

我建议先阅读这个与 1 个生产者和 1 个消费者一起工作的消费者-生产者队列的无锁实现：无锁的多线程单生产者单消费者： http://www.codeproject.com/KB/threads/LockFree.aspx#heading0005

我还建议阅读一些有关 volatile 的内容： http://www.drdobbs.com/cpp/212701484

Herb Sutter 写了一篇很好的文章，提醒你编写这种代码的危险：http://www.drdobbs.com/cpp/210600279;jsessionid=ZSUN3G3VXJM0BQE1GHRSKHWATMY32JVN?pgno=2

最后我建议阅读这篇文章以及另一个无锁队列：http://www.drdobbs.com/architecture-and-design/210604448

我还应该指出竞争条件并使用缓存大小的内存块来存储您写入的空间并将其与您正在读取的缓存行分开以避免一个线程强制另一个线程重新获取数据从主内存。

希望有帮助

【讨论】：

【解决方案2】：

紧密循环会影响多线程。听起来您的第一个线程运行得足够快，以至于第二个线程甚至没有机会启动。注意，如果发生这种情况，那么顺序解决方案是最有效的。 :)

由于您的生产者/消费者布局有些复杂，我假设您在真实数据中看不到这种行为。

虽然您可以通过添加带有非零参数的Thread.Sleep（参见Joe Duffy's blog entry on why this works）或忽略它来解决此问题，但更好的解决方案是限制第一个生产者/消费者队列的大小。这将只允许第一个线程产生一定数量的小部件，然后阻塞，直到管道的其余部分有机会启动。

.NET 4.0 BlockingCollection<T> 允许您指定最大大小。 Microsoft Rx library 已将其反向移植到 .NET 3.5，因此您可以根据需要使用它。（我确实建议这样做，而不是使用内部解决方案）。

【讨论】：

“听起来您的第一个线程运行得足够快，以至于第二个线程甚至没有机会启动。”百万美元的问题是：为什么会这样？为什么一个线程的运行速度会阻止另一个线程运行？那永远不应该发生！
请记住，其中涉及到一个任务调度程序。一个线程得到它的时间片；在某些情况下甚至更多。第一个线程实际运行了多长时间？
它以 30 毫秒的时间间隔运行 100 万条记录（几分钟）。线程调度程序应该循环调度具有相同优先级的其他线程。通常，只有持有竞争锁才会导致这种行为。但是我的第一个线程只有大约一半的时间有锁。如果有 100 万条记录，那么肯定至少有一次线程会在锁外挂起，从而允许下一个线程运行。
你没有搞乱线程优先级，是吗？所有这些线程都处于默认优先级，对吗？
是的，它们都具有相同的优先级（不涉及优先级）。我想“也许第一个线程，一个前台线程，比第二个线程，一个后台线程池线程具有更高的优先级。”基于此，我将第一个线程更改为后台线程（我仍然没有明确地触及优先级）。这种调整并没有改变结果。