PFX ConcurrentQueue - есть ли способ удалить определенный элемент из очереди

Question

У меня есть приложение, в котором есть ConcurrentQueue элементов, имеющих свойство ID, и ConcurrentQueue задач для каждого элемента, элементы очереди выглядят следующим образом:

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

и сама очередь выглядит следующим образом:

ConcurrentQueue<QueueItem> itemQueue;

У меня есть один поток, выполняющий foreach над itemQueue, удаляющий элемент из каждой очереди и выполняющий работу над ним:

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.

Моя проблема возникает, когда у меня больше нет рабочих элементов в QueueItem - я бы хотел удалить этот QueueItem из itemQueue - что-то вроде...

  if (queueItem.workItemQueue.TryDequeue(out workItem))
    doWork(workItem);
  else
    itemQueue.TryRemove(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

Это может сработать не для всех, но ниже приведено решение, которое я придумал для удаления элемента из параллельной очереди, поскольку это первый результат Google, я подумал, что оставлю свое решение позади.

Что я сделал, так это временно заменил рабочую очередь пустой, преобразовал исходную в список и удалил элемент (ы), затем создал новую очередь из измененного списка и вернул ее обратно.

В коде (извините, это 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

Очереди предназначены, когда вы хотите обрабатывать элементы в стиле FIFO, стеки для LIFO.Существует также concurrentdictionary и concurrentbag.Убедитесь, что очередь - это действительно то, что вам нужно.Я не думаю, что я бы когда-нибудь сделал foreach для concurrentqueue .

То, что вы, вероятно, хотите, - это единая очередь для ваших рабочих элементов (пусть они используют общий интерфейс и создают очередь в интерфейсе, интерфейс должен предоставлять унаследованный тип, к которому он может быть преобразован позже, если это необходимо).Если workitems принадлежат родительскому элементу, то можно использовать свойство, которое будет содержать ключ к родительскому элементу (рассмотрим GUID для ключа), а родительский элемент можно сохранить в concurrentdictionary и ссылаться на него / удалять по мере необходимости.

Если вы должны сделать это так, как у вас есть, подумайте о добавлении флага.затем вы можете пометить элемент в itemqueue как "закрытый" или что-то еще, так что, когда он будет удален из очереди, он будет проигнорирован.