等待阻塞收集（队列）在 C# 中减小大小答案

【问题标题】：Wait for blocking collection (queue) to decrease in size in C#等待阻塞收集（队列）在 C# 中减小大小
【发布时间】：2013-02-21 14:33:52
【问题描述】：

我正在使用以下工作流程处理项目：

第一部分：

事件异步到达并在阻塞队列中排队，我们称之为Q1
线程从该队列中获取下一个可用项目
项目最终并行运行 {N} 个任务
每个任务将其结果排入第二个队列，我们称之为Q2。
当项目处理完成时，从队列中读取下一个项目。

第二部分：

另一个线程一次读取 Q2 一个对象并处理结果

因此，这里的问题是，第一个队列中的每个项目最终都会并行运行大量任务，并且每个任务都将其结果排队。第二个队列必须按顺序处理，但一次一个项目，它会被淹没。

我的问题

我需要一种机制，使线程处理 Q1 等到 Q2 中的项目数低于特定阈值。实现这一目标的最佳方法是什么？有什么办法可以让事件驱动解决方案而不是轮询解决方案？

【问题讨论】：

标签： c# .net multithreading queue event-driven

【解决方案1】：

我假设您在偶尔发生的洪水中收到数据，在长时间的干旱中，第二季度可以赶上。您是否考虑过通过为这些任务使用有限的线程池来简单地限制从第一季度产生的并发线程的数量？

如果作业大小在到达时很容易确定，我怀疑您可以从多个线程池中受益。您可以拥有少量线程来处理大型作业，而大量线程准备好处理小型作业。即使是第三个中间队列也可能是有益的。

【讨论】：

我们已经使用了线程池。问题是很难判断要使用多少线程。有时我们会收到 80k 个很小的事件并且系统很好（最后一个队列不会堆积，因为结果集很小/快速上传到我们的分布式缓存）。其他时候，我们会收到 100 个事件，它会挂起，因为结果集很大，上传需要很长时间。因此，当结果集很小时，完全限制池会使它运行得更糟，但是当它们太大时确实解决了泛滥的问题。 +1 虽然这是一个很好的解决方案，但我提供了 :)

【解决方案2】：

您的问题似乎是 TPL Dataflow 库解决的完美示例。如果你愿意尝试，下面是它的工作原理（当然这是非常简单的例子）：

TransformBlock<int, bool> transform = new TransformBlock<int, bool>(i => i > 5 ? true : false,
            new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 4 });
ActionBlock<bool> resultBlock = new ActionBlock<bool>(b => Console.WriteLine("My result is : " + b),
            new ExecutionDataflowBlockOptions { BoundedCapacity = 10 });
transform.LinkTo(resultBlock);

您正在定义一个转换块，它将进行您的转换（这与您的 Q1 一样），您可以将它的并行度级别设置为您想要使用的任务数。

然后，您正在创建第二个块（作为您的Q2 工作），它将设置BoundedCapacity，它同步处理每条消息，为每个元素调用一个操作。此块可以被任何其他块替换，例如 BufferBlock，这将允许您按需从中进行轮询。

【讨论】：

我现在没时间玩这个，但看起来很有趣；我会再试一次:) 谢谢。

【解决方案3】：

您可以在 Q2 中使用 BlockingCollection<T>，而不是使用 Queue<T>。如果您设置了它的BoundedCapacity，那么当达到容量时，对Q2.Add() 的调用将被阻塞。这将自动限制 Q1 的处理，因为如果 N 个任务无法添加到最终队列中，它们将开始阻塞。

【讨论】：

酷；我已经在这里使用了阻塞集合，所以这是一个小改动:)
我会尝试并接受它是否有效。这可能是一个很好的解决方案，因为它会阻止我的任务处理 {N} 个线程返回，直到它们可以将结果排队，这将首先阻止更多事件被处理:)