PFX ConcurrentQueue - 有没有办法从队列中删除特定项目

Question

我有一个应用程序，它有一个具有 ID 属性的项目的 ConcurrentQueue 和每个项目的任务的 ConcurrentQueue，队列项目看起来像：

class QueueItem {
  public int ID { get; set; }
  public ConcurrentQueue<WorkItem> workItemQueue { get; set; }
}

队列本身看起来像：

ConcurrentQueue<QueueItem> itemQueue;

我有一个线程在 itemQueue 上执行 foreach，将一个每个队列中的项目并对其进行工作：

foreach(var queueItem in itemQueue) {
  WorkItem workItem;
  if (queueItem.workItemQueue.TryDequeue(out workItem))
    doWork(workItem);
  else
    // no more workItems for this queueItem
}

我正在使用 ConcurrentQueues，因为我有一个单独的线程可能将queueItems添加到itemQueue，并将workItems添加到每个workItemQueue。

当我在队列项目中没有更多的工作项目时，我的问题就出现了 - 我想从项目队列中删除该队列项目 - 就像......

  if (queueItem.workItemQueue.TryDequeue(out workItem))
    doWork(workItem);
  else
    itemQueue.TryRemove(queueItem);

但我找不到一种方法可以轻松做到这一点。 我想出的方法是使每个 QueueItem 出队，然后在 workItemQueue 中仍有 WorkItems 时将其入队：

for (int i = 0; i < itemQueue.Count; i++) {
  QueueItem item;
  itemQueue.TryDequeue(out queueItem);
  if (queueItem.workItemQueue.TryDequeue(out workItem)) {
    itemQueue.Enqueue(queueItem);
    doWork(workItem);
  }
  else
    break;
}

是否有更好的方法来使用 PFX ConcurrentQueue 完成我想要的任务，或者这是执行此操作的合理方法，我应该使用自定义并发队列/列表实现还是我遗漏了一些东西？

Answer 1

一般来说，没有有效的方法从队列中删除特定项目。 它们通常具有 O(1) 队列和出列，但 O(n) 删除，这就是您的实现所做的。

一种替代结构称为 LinkedHashMap。 查看 Java 实现 如果你感兴趣。

它本质上是一个哈希表和一个链表，允许 O(1) 排队、出队和删除。

这尚未在 .Net 中实现，但网络上有一些实现。

现在的问题是，为什么 itemQueue 是一个队列？ 从您的代码示例中，您永远不会将任何内容放入队列或出列（除了解决“删除”问题）。 我怀疑如果使用更合适的数据结构，您的问题可以简化。 您能举例说明哪些其他代码片段访问 itemQueue 吗？

Answer 2

这可能并不适合所有人，但以下是我想出的从并发队列中删除项目的解决方案，因为这是第一个谷歌结果，我想我会留下我的解决方案。

我所做的就是暂时用空队列替换工作队列，将原始队列转换为列表并删除项目，然后从修改后的列表中创建一个新队列并将其放回原处。

在代码中（抱歉，这是 VB.net 而不是 C#）：

Dim found As Boolean = False
//'Steal the queue for a second, wrap the rest in a try-finally block to make sure we give it back
Dim theCommandQueue = Interlocked.Exchange(_commandQueue, New ConcurrentQueue(Of Command))
Try
    Dim cmdList = theCommandQueue.ToList()
    For Each item In cmdList
        If item Is whateverYouAreLookingFor Then
            cmdList.Remove(item)
            found = True
        End If
    Next
    //'If we found the item(s) we were looking for, create a new queue from the modified list.
    If found Then
        theCommandQueue = New ConcurrentQueue(Of Command)(cmdList)
    End If
Finally
    //'always put the queue back where we found it
    Interlocked.Exchange(_commandQueue, theCommandQueue)
End Try

旁白：这是我的第一个答案，因此请随意提出一些编辑建议和/或编辑我的答案。

Answer 3

当您想要以先进先出的方式处理项目时，队列意味着后进先出的堆栈。 还有一个并发字典和一个并发包。 确保队列确实是您想要的。 我认为我不会在并发队列上执行 foreach 操作。

您可能想要的是工作项的单个队列（让它们使用公共接口并在接口上创建一个队列，该接口应该公开继承的类型，以后如果需要的话可以将其重新转换为该继承类型）。 如果工作项属于父项，则可以使用一个属性来保存父项的密钥（考虑密钥的 GUID），并且可以将父项保留在并发字典中并根据需要引用/删除。

如果您必须按照现有方式进行操作，请考虑添加一个标志。 然后，您可以将 itemqueue 中的项目标记为“已关闭”或其他任何内容，这样当它出列时，它将被忽略。