PFX ConcurrentQueue - est-il un moyen de supprimer un élément spécifique de la file d'attente

Question

J'ai une application qui a une ConcurrentQueue d'éléments qui ont une propriété ID et une ConcurrentQueue de tâches pour chaque élément, les éléments de file d'attente ressemblent:

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

et la file d'attente se ressemble:

ConcurrentQueue<QueueItem> itemQueue; 

J'ai un fil faire un foreach sur la itemQueue, deQueueing un élément de chaque file d'attente et faire un travail sur elle:

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

J'utilise ConcurrentQueues parce que j'ai un thread séparé qui ajoute potentiellement des queueItems à l'itemQueue et qui ajoute des workItems à chaque workItemQueue.

Mon problème vient quand je n'ai pas plus WorkItems dans un queueItem - Je voudrais supprimer cette queueItem du itemQueue - quelque chose comme ...

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

... mais je ne peux pas trouver un moyen de le faire facilement. La façon dont je suis venu avec est de dequeue chaque QueueItem puis Enqueue s'il y a WorkItems encore dans le workItemQueue:

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; 
} 

Y at-il une meilleure façon d'accomplir ce que je veux à l'aide du PFX ConcurrentQueue, ou est-ce Si vous utilisez une implémentation de liste/file d'attente simultanée personnalisée ou que je manque quelque chose, est-ce une manière raisonnable de le faire?

Answer 1

En général, il n'existe aucun moyen efficace de supprimer des éléments spécifiques des files d'attente. Ils ont généralement la file d'attente O (1) et les dequeues, mais O (n) supprime, ce que votre implémentation fait.

Une structure alternative est appelée LinkedHashMap. Jetez un oeil à la Java implementation si vous êtes intéressé.

Il s'agit essentiellement d'une table de hachage et une liste chaînée, qui permet d'O (1) file d'attente, dequeue et remove.

Cela n'est pas encore implémenté dans .Net, mais il existe quelques implémentations flottant sur le Web. Maintenant, la question est: pourquoi itemQueue est une file d'attente? À partir de vos exemples de code, vous n'en mettez jamais en file d'attente ou n'en supprimez rien (sauf pour naviguer autour du problème Supprimer). Je soupçonne que votre problème pourrait être simplifié si une structure de données plus appropriée est utilisée. Pourriez-vous donner des exemples sur les autres éléments d'accès au code itemQueue?

Answer 2

Cela peut ne pas fonctionner pour tout le monde, mais voici la solution que j'ai trouvée pour supprimer un élément d'une file d'attente simultanée, puisque c'est le premier résultat google, je pensais laisser ma solution. Ce que j'ai fait était de remplacer temporairement la file d'attente de travail par un vide, de convertir l'original en liste et de supprimer le (s) élément (s), puis de créer une nouvelle file dans la liste modifiée et de la remettre.

Dans le code (désolé c'est plutôt 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 

En plus: Ceci est ma première réponse, alors ne hésitez pas à mettre en place des conseils de montage et/ou modifier ma réponse.

Answer 3

Les files d'attente sont utilisées lorsque vous souhaitez gérer des éléments dans un style FIFO, Stacks for LIFO. Il y a aussi un concurrentdictionary et un concurrentbag. Assurez-vous qu'une file d'attente est réellement ce que vous voulez. Je ne pense pas que je ferais jamais un foreach sur une file d'attente concurrente. Ce que vous voudrez probablement est une seule file d'attente pour vos éléments de travail (demandez-leur d'utiliser une interface commune et de faire une file d'attente sur l'interface, l'interface devrait exposer le type hérité auquel elle peut être refondue si nécessaire). Si les éléments de travail appartiennent à un parent, alors une propriété peut être utilisée qui contiendra une clé pour le parent (considérez un GUID pour la clé), et le parent peut être conservé dans un paralleldictionary et référencé/supprimé si nécessaire.

Si vous devez le faire comme vous l'avez, pensez à ajouter un drapeau. Vous pouvez ensuite marquer l'item dans la file d'item comme étant 'closed' ou autre, de sorte que quand il est retiré, il sera ignoré.