线程安全的缓冲可观察优先级队列？

Question

我正在编写一个程序，其中一个线程需要将项目推送到队列中，而一个或多个线程则从队列中弹出项目并处理它们。为了避免内存不足，我希望在队列满时生产者线程能够休眠。有些项目比其他项目具有更高的优先级，因此我希望这些项目首先被处理。如果项目具有相同的优先级，则希望最先添加的项目先被处理。

我想在WPF DataGrid中显示前100个项目左右，因此它也需要由UI线程访问。如果它能通知UI线程已经更新，即实现IObservable，那就太好了。

是否有一个容器类可以完成所有这些操作？

额外加分的是，我相信在入队和出队时并不需要锁定整个队列。

.NET 4 的实现是可以的。

- mpen

---

BlockingCollection http://msdn.microsoft.com/en-us/library/dd997371.aspx 听起来很有希望... 但它没有提到任何关于优先级的内容。它确实说它可以封装任何实现了 IProducerConsumerCollection 接口的东西... 是否有一个优先级队列实现了这个接口呢？ - mpen

请参考此示例：http://msdn.microsoft.com/zh-cn/library/dd460690.aspx - Sean Fausett

顺便提一下，任务可以转换为 Rx observables。 ;) - Sean Fausett

Answer 1

如果你正在寻找一个容器，那么很遗憾，你必须自己实现。在优先级方面要小心——排序速度会非常缓慢。我的做法是实现一个队列类，内部使用多个数组（每个优先级一个数组——低、中、高编码）。这样我就不需要排序了。在可以的情况下避免锁定（多核假定），并选择自旋锁（.NET 4.0），它们在队列场景中更快/承载更少的开销。

- TomTom

---

我永远不会在每次插入时重新排序整个集合...那太疯狂了。我会像你建议的那样使用多个数组（或队列），或者某种树形结构。 - mpen

树结构比多个队列慢（但如果可能的优先级数量很高，则更有效 - 这很少有意义）。树的重新平衡在价格方面是非常昂贵的，而且在队列中你也不需要搜索它们。 - TomTom

Answer 2

过去我所做的是将多个ConcurrentQueue<T>集合包装成一个集合，有点像TomTom建议的那样。当你打算要有的优先级数量较少时，这是相当合理的。例如，在某些情况下，仅拥有高优先级和低优先级可能就足够了。然后您的TryDequeue方法看起来就像这样：

public bool TryDequeue(out T item)
{
    return _highItems.TryDequeue(out item) || _lowItems.TryDequeue(out item);
}

这并不是对你问题的全面回答，但或许可以帮助你起步。

- Dan Tao

---

我不想限制可能优先级的数量。我想使用计算出的值作为优先级，但是事先我并不知道范围。 - mpen

@Mark：有时人们会在内部使用类似于 SortedList<TPriority, Queue<T>> 的东西来实现这个功能；但正如TomTomTom已经提到的，无限制的优先级成本最终可能不值得。那么怎么办呢？可以妥协一下：计算一个值，根据它在某个范围内的位置将其转换为预定义的优先级。 - Dan Tao

是的...我在想也许那就是我应该做的。我讨厌对事物施加任意限制，但如果不这样做，我想我会为自己创造太多工作。当我意识到我必须锁定结构体以便在入队和出队时管理键时，我已经完成了使用SortedDictionary<TPriority, ConcurrentQueue<TValue>>实现的一半，这会减少一些并发优势 :( - mpen

Answer 3

如果您使用的是.NET 4，您应该认真考虑使用任务并行库和一个自定义调度程序，例如示例QueuedTaskScheduler。我不确定它是否符合您的所有要求，但这将是一个很好的开始。

- Sean Fausett

---

我的答案可能有偏差，但我强烈反对认为它不能解决问题。专用线程上的数据处理实际上是通过集合中数据的排序来有效地安排的。任务是一种更高级的抽象，可以封装数据及其处理过程，而不必直接处理线程——这有许多好处。在这种情况下，调度可以通过任务调度程序来实现。 - Sean Fausett