Création d'une File d'attente de blocage<T> dans .NET?

J'ai un scénario où j'ai plusieurs threads ajout d'une file d'attente et plusieurs threads de lecture à partir de la même file d'attente. Si la file d'attente atteint une taille spécifique tous les threads qui sont le remplissage de la file d'attente sera bloqué sur ajouter jusqu'à un point est retiré de la file d'attente.

La solution ci-dessous est ce que je suis en ce moment et ma question est: Comment cela peut-il être amélioré? Est-ce un objet qui permet déjà ce comportement dans la BCL que je devrais utiliser?

internal class BlockingCollection<T> : CollectionBase, IEnumerable
{
//todo: might be worth changing this into a proper QUEUE
private AutoResetEvent _FullEvent = new AutoResetEvent(false);
internal T this[int i]
{
get { return (T) List[i]; }
}
private int _MaxSize;
internal int MaxSize
{
get { return _MaxSize; }
set
{
_MaxSize = value;
checkSize();
}
}
internal BlockingCollection(int maxSize)
{
MaxSize = maxSize;
}
internal void Add(T item)
{
Trace.WriteLine(string.Format("BlockingCollection add waiting: {0}", Thread.CurrentThread.ManagedThreadId));
_FullEvent.WaitOne();
List.Add(item);
Trace.WriteLine(string.Format("BlockingCollection item added: {0}", Thread.CurrentThread.ManagedThreadId));
checkSize();
}
internal void Remove(T item)
{
lock (List)
{
List.Remove(item);
}
Trace.WriteLine(string.Format("BlockingCollection item removed: {0}", Thread.CurrentThread.ManagedThreadId));
}
protected override void OnRemoveComplete(int index, object value)
{
checkSize();
base.OnRemoveComplete(index, value);
}
internal new IEnumerator GetEnumerator()
{
return List.GetEnumerator();
}
private void checkSize()
{
if (Count < MaxSize)
{
Trace.WriteLine(string.Format("BlockingCollection FullEvent set: {0}", Thread.CurrentThread.ManagedThreadId));
_FullEvent.Set();
}
else
{
Trace.WriteLine(string.Format("BlockingCollection FullEvent reset: {0}", Thread.CurrentThread.ManagedThreadId));
_FullEvent.Reset();
}
}
}
  • .Net comment construit-dans les classes pour aider à ce scénario. La plupart des réponses énumérées ici sont obsolètes. Voir les plus récentes réponses au bas. Regardez dans les thread-safe blocage des collections. Les réponses peuvent être obsolètes, mais c'est toujours une bonne question!
  • Je pense que c'est toujours une bonne idée d'apprendre sur le Moniteur.Wait/Impulsion/PulseAll même si nous avons des nouvelles concurrentes des classes dans .NET.
  • D'accord avec @thewpfguy. Vous aurez envie de comprendre la base des mécanismes de verrouillage de derrière les coulisses. Également intéressant de noter que les Systèmes.Les Collections.Simultanées n'existait pas jusqu'en avril 2010, et alors seulement dans Visual Studio 2010 et au-dessus. Définitivement pas une option pour le VS2008 tenir les aboutissants...
  • Si vous lisez ceci maintenant, jetez un oeil au Système.Le filetage.Les canaux multi-écrivain/multi-lecteur, borné, éventuellement le blocage de la mise en œuvre de ce pour .NET de Base et .NET Standard.
InformationsquelleAutor Eric Schoonover | 2009-02-09
  1. 198

    Qui a l'air très dangereux (très peu de synchronisation); comment quelque chose comme:

    class SizeQueue<T>
{
private readonly Queue<T> queue = new Queue<T>();
private readonly int maxSize;
public SizeQueue(int maxSize) { this.maxSize = maxSize; }
public void Enqueue(T item)
{
lock (queue)
{
while (queue.Count >= maxSize)
{
Monitor.Wait(queue);
}
queue.Enqueue(item);
if (queue.Count == 1)
{
//wake up any blocked dequeue
Monitor.PulseAll(queue);
}
}
}
public T Dequeue()
{
lock (queue)
{
while (queue.Count == 0)
{
Monitor.Wait(queue);
}
T item = queue.Dequeue();
if (queue.Count == maxSize - 1)
{
//wake up any blocked enqueue
Monitor.PulseAll(queue);
}
return item;
}
}
}

    (edit)

    En réalité, vous voulez un moyen de fermer la file d'attente afin que les lecteurs commencer à sortir proprement - peut-être quelque chose comme un bool drapeau - le cas échéant, une vide la file d'attente retourne juste (plutôt que le blocage):

    bool closing;
public void Close()
{
lock(queue)
{
closing = true;
Monitor.PulseAll(queue);
}
}
public bool TryDequeue(out T value)
{
lock (queue)
{
while (queue.Count == 0)
{
if (closing)
{
value = default(T);
return false;
}
Monitor.Wait(queue);
}
value = queue.Dequeue();
if (queue.Count == maxSize - 1)
{
//wake up any blocked enqueue
Monitor.PulseAll(queue);
}
return true;
}
}
    • Merci Marc, super réponse. Je vais lui donner un peu d'heures, de sorte que si quelqu'un d'autre a des bonnes idées, ils ont une chance pour la accepté. C'est super bien, merci!
    • Ne serait-il pas également nécessaire de Surveiller.PulseAll en cas de proximité?
    • Comment sur la modification de l'attendre à un WaitAny et en le passant dans un résilier waithandle sur la construction ...
    • pour quel but? Au moment où il utilise le Moniteur, ce qui est le moins cher de tous le verrouillage de constructions, car il n'a pas besoin d'aller à l'OS. Un booléen et un Moniteur est un peu moins cher qu'un waithandle...
    • Yerp, votre mise en œuvre ici est solide, facile à comprendre et rapide. Seul avantage avec un WaitAny type d'approche est que vous pourriez avoir un seul événement qui déclenche l'arrêt, sans avoir besoin d'une méthode Close ... Il peut être utile parfois.
    • une autre chose que je voudrais examiner ici, c'est sans doute la taille de la file d'attente IDisposable. Aussi, ce qui se passe si votre file d'attente de l'objet est hors de portée alors que les threads sont bloqués. Il serait sans doute un bougre à déboguer.
    • une optimisation, si vous vous attendiez à la file d'attente de toujours atteindre la capacité, serait de passer le maxSize valeur dans le constructeur de la File d'attente<T>. Vous pouvez ajouter un autre constructeur de votre classe pour accueillir les pour que.
    • Pourquoi SizeQueue, pourquoi ne pas FixedSizeQueue?
    • Si vous bloquez sur la file d'attente, ne sont pas que de prévenir efficacement les autres threads de jouer avec elle? Par exemple, ajouter, d'abord, il bloque la file d'attente, et puis si la file est pleine, elle attend ensuite pour la file d'attente pour être signalé, mais comment peut-il être signalée si aucun autre thread ne peut le verrouiller en premier? Il doit y avoir quelque chose, je ne suis pas ici...
    • il libère le verrou(s) au cours de Wait, de sorte que les autres threads peuvent l'acquérir. Il libère le verrou(s) quand il se réveille.
    • Nice, comme je l'ai dit, il y avait quelque chose que je n'étais pas 🙂 ce Qui me donne envie de revoir certains de mes thread-code....
    • Ce qui se passe lorsqu'un thread se réveille et que le verrou est déjà pris par quelqu'un d'autre, il permet de bloquer en attente pour le verrouillage de devenir libre et de prétendre ensuite il?
    • Effectivement je peux penser à de nombreuses questions autour de ce sujet, il existe un lien de quelque matériel que je peux lire? Si pas, je vais essayer de résumer quelques questions dans un post de mon propre ici.
    • quand il est pulsé il se déplace de la de couchage-file d'attente pour le prêt-à-queue (comme si c'était tout à l'heure en essayant de prendre le verrou). Lorsque le thread qui pulsé libère le verrou, le suivant fil dans le prêt-à-queue (qui peut-être aucun de ceux-ci) obtient un aller. Et ainsi de suite. MSDN peuvent aider, mais... de Jon pages de bénéficier d'une couverture trop.
    • remarque le menu déroulant en bas pour les 17 sujets: yoda.arachsys.com/csharp/threads
    • Ne serait pas d'Impulsion au lieu de PulseAll bien fonctionner dans la File/file d'attente?
    • peut-être, mais alors vous en grande partie à savoir le thread est en attente. Et dans le cas de la traction de la file d'attente vers le bas vous activement voulez pour réveiller tous les travailleurs. Dans la plupart des cas, il n'existe qu'un travailleur, en faisant un point discutable. Dans les cas plus complexes, vous avez assez de soucis, sans stress, sans plus de cela - vous pouvez avec soin l'ajouter, je suppose.
    • En réponse à ma propre question: non, l'Impulsion n'est pas suffisant car il se réveille dans un thread unique et nous ne savons pas si ce fil est dans la mise en File d'attente ou de Retirer de la méthode.
    • Dans le cas où quelqu'un se soucie j'ai choisi de ne pas avoir un maximum de taille de file d'attente. Ainsi, la mise en file d'appel n'blocs de signification file d'attente n'a pas besoin d'appel d'impulsion et de mise en file d'attente peuvent utiliser l'Impulsion au lieu de PulseAll depuis seulement file d'attente des appels en attente. Notez que lors de la fermeture de la file d'attente PulseAll est bien sûr nécessaire. Merci Marc pour la réponse rapide.
    • Je sais que c'est un vieux post, mais j'ai une petite question. Lorsque Enqueue appels PulseAll il libère le moniteur à chaque thread en attente et la course pour verrouiller le moniteur de l'objet de nouveau, est-il possible que si un autre écrivain va gagner et de dire que la file d'attente est pleine à ce moment, il va entrer Monitor.Wait de nouveau et puis nous nous retrouvons avec un blocage?
    • ce n'est pas une impasse; une Attente libère le verrou à la suivante dans la liste, donc à un certain point un lecteur va la chercher, et réveillera que l'écrivain quand il a fait de l'espace
    • ahh, j'oublie que le .Wait libère le verrou!! Le sens parfait maintenant. Thx.
    • bool TryEnqueue(T item) est aussi nécessaire à la sortie correctement les threads bloqués une fois queue.Count atteint maxSize. Les éléments de queue sera pas être consommés, mais ne peut rien faire beaucoup parler.
    • il a certainement pourrait être nécessaire, si vous avez besoin d'une méthode qui ajoute si il ya de la place, le reste n'a pas d'ajout et retourne immédiatement. Je n'ai pas besoin de ça 🙂
    • Basé sur votre classe, j'ai eu à mettre en œuvre un TryDequeueWhere, à retirer un objet en fonction de certains critères. Pourriez-vous avoir un coup d'oeil à ceci : stackoverflow.com/q/47174748/1959238

    InformationsquelleAutor Marc Gravell
  2. 55

    Utilisation .net 4 BlockingCollection, de mise en file d'utiliser la fonction Ajouter(), à la résorption de l'utilisation de(). Il utilise en interne non bloquant ConcurrentQueue. Plus d'info ici Rapide et Meilleur Producteur/consommateur file d'attente de la technique BlockingCollection vs simultanées de la File d'attente

    InformationsquelleAutor xhafan
  3. 14

    "Comment cela peut-il être amélioré?"

    Bien, vous avez besoin de regarder toutes les méthodes de votre classe et de demander ce qui se passerait si un autre thread a été simultanément l'appel de cette méthode ou toute autre méthode. Par exemple, vous avez mis un verrou dans la méthode Remove, mais pas dans la méthode Add. Qu'advient-il si un thread Ajoute en même temps qu'un autre thread Supprime? De mauvaises choses.

    Considèrent également qu'une méthode peut retourner une deuxième objet qui permet d'accéder à la première de l'objet de données internes - par exemple, GetEnumerator. Imaginez un fil traverse que les agents recenseurs, un autre thread, c'est de modifier la liste en même temps. Pas bon.

    Une bonne règle de base est de faire le plus simple pour obtenir le droit de couper le nombre de méthodes dans la classe au minimum absolu.

    En particulier, ne pas hériter d'une autre classe de conteneur, parce que vous allez vous exposer l'ensemble de cette classe de méthodes, fournissant un moyen pour l'appelant de corrompre la base de données interne, ou de voir partiellement effectuer les modifications apportées aux données (tout aussi mauvais, parce que les données s'affiche corrompu à ce moment). Masquer tous les détails et de se montrer impitoyable à propos de la façon dont vous autorisez l'accès à eux.

    J'aimerais vous conseillons fortement d'utiliser de solutions prêtes à l'emploi - un livre sur le filetage ou utiliser la 3ème partie de la bibliothèque. Autrement, compte tenu de ce que vous tentez, vous allez être le débogage du code pour une longue période de temps.

    Aussi, ne serait-il pas plus logique de Supprimer le retour d'un élément (par exemple, celui qui a été ajouté en premier, car il est une file d'attente), plutôt que de l'appelant de choisir un élément spécifique? Et lorsque la file d'attente est vide, peut-être Supprimer doit également bloquer.

    Mise à jour: la réponse de Marc en fait met en œuvre toutes ces suggestions! 🙂 Mais je vais laisser cela ici comme il peut être utile de comprendre pourquoi sa version est une amélioration.

    • J'apprécie les explications détaillées. Merci!
    InformationsquelleAutor Daniel Earwicker
  4. 10

    Vous pouvez utiliser le BlockingCollection et ConcurrentQueue dans le Système.Les Collections.Simultanée De L'Espace De Noms

     public class ProducerConsumerQueue<T> : BlockingCollection<T>
{
///<summary>
///Initializes a new instance of the ProducerConsumerQueue, Use Add and TryAdd for Enqueue and TryEnqueue and Take and TryTake for Dequeue and TryDequeue functionality
///</summary>
public ProducerConsumerQueue()  
: base(new ConcurrentQueue<T>())
{
}
///<summary>
///Initializes a new instance of the ProducerConsumerQueue, Use Add and TryAdd for Enqueue and TryEnqueue and Take and TryTake for Dequeue and TryDequeue functionality
///</summary>
///<param name="maxSize"></param>
public ProducerConsumerQueue(int maxSize)
: base(new ConcurrentQueue<T>(), maxSize)
{
}
}
    • BlockingCollection par défaut de la File d'attente. Donc, je ne pense pas que ce soit nécessaire.
    • Ne BlockingCollection préserver l'ordre comme une file d'attente?
    • Oui, quand il est initialisé avec un ConcurrentQueue
    InformationsquelleAutor Andreas
  5. 6

    Je viens juste d'en cela de l'aide du Réactif Extensions et de rappeler à cette question:

    public class BlockingQueue<T>
{
private readonly Subject<T> _queue;
private readonly IEnumerator<T> _enumerator;
private readonly object _sync = new object();
public BlockingQueue()
{
_queue = new Subject<T>();
_enumerator = _queue.GetEnumerator();
}
public void Enqueue(T item)
{
lock (_sync)
{
_queue.OnNext(item);
}
}
public T Dequeue()
{
_enumerator.MoveNext();
return _enumerator.Current;
}
}

    Pas nécessairement entièrement sûr, mais très simple.

    • Ce qui est Sujet<t>? Je n'ai pas de résolution pour son espace de noms.
    • Il fait partie de la Réactif Extensions.
    • Pas de réponse. Ceci ne répond pas à la question.
    InformationsquelleAutor Mark Rendle
  6. 5

    C'est ce que je suis venu op pour une thread-safe délimitée bloque la file d'attente. 

    using System;
using System.Collections.Generic;
using System.Text;
using System.Threading;
public class BlockingBuffer<T>
{
private Object t_lock;
private Semaphore sema_NotEmpty;
private Semaphore sema_NotFull;
private T[] buf;
private int getFromIndex;
private int putToIndex;
private int size;
private int numItems;
public BlockingBuffer(int Capacity)
{
if (Capacity <= 0)
throw new ArgumentOutOfRangeException("Capacity must be larger than 0");
t_lock = new Object();
buf = new T[Capacity];
sema_NotEmpty = new Semaphore(0, Capacity);
sema_NotFull = new Semaphore(Capacity, Capacity);
getFromIndex = 0;
putToIndex = 0;
size = Capacity;
numItems = 0;
}
public void put(T item)
{
sema_NotFull.WaitOne();
lock (t_lock)
{
while (numItems == size)
{
Monitor.Pulse(t_lock);
Monitor.Wait(t_lock);
}
buf[putToIndex++] = item;
if (putToIndex == size)
putToIndex = 0;
numItems++;
Monitor.Pulse(t_lock);
}
sema_NotEmpty.Release();
}
public T take()
{
T item;
sema_NotEmpty.WaitOne();
lock (t_lock)
{
while (numItems == 0)
{
Monitor.Pulse(t_lock);
Monitor.Wait(t_lock);
}
item = buf[getFromIndex++];
if (getFromIndex == size)
getFromIndex = 0;
numItems--;
Monitor.Pulse(t_lock);
}
sema_NotFull.Release();
return item;
}
}
    • Pourriez-vous donner quelques exemples de code de la façon dont j'avais la queue du fil fonctions à l'aide de cette bibliothèque, y compris comment j'avais instancier cette classe?
    • Cette question/réponse est un peu daté. Vous devriez regarder le Système.Les Collections.Simultanée de l'espace de noms pour le blocage de la file d'attente de soutien.
    InformationsquelleAutor Kevin
  7. 2

    Je n'ai pas encore totalement exploré la TPL mais ils pourraient avoir quelque chose qui correspond à vos besoins, ou à tout le moins, certains Réflecteur de fourrage pour accrocher l'inspiration.

    Espère que ça aide.

    • Merci pour le lien, je vais regarder de plus près à la TPL.
    InformationsquelleAutor TheMissingLINQ
  8. 0

    Bien, vous pourriez regarder System.Threading.Semaphore classe. Autre que cela - pas, vous avez à faire vous-même. Autant que je sache, il n'y a pas une telle collection intégrée.

    • J'ai regardé pour la limitation du nombre de threads qui sont l'accès à une ressource, mais elle ne vous permet pas de bloquer tous les accès à une ressource basée sur une condition (comme Collection.Le comte). Autant que je sache, de toute façon
    • Eh bien, vous n'avez qu'une partie yoursef, tout comme vous le faites maintenant. Simplement au lieu de MaxSize et _FullEvent vous avez le Sémaphore, qui vous initialisez avec le droit de compter dans le constructeur. Ensuite, lors de chaque Ajout/suppression de vous appeler WaitForOne() ou Release().
    • Il n'est pas très différent de ce que vous avez maintenant. Juste plus simple à mon humble avis.
    • Pouvez-vous me donner un exemple montrant ce travail? Je ne vois pas comment ajuster la taille d'un Semaphor de manière dynamique ce qui ce scénario exige. Depuis que vous avez à être en mesure de bloquer toutes les ressources que si la file d'attente est pleine.
    • Ahh, la modification de la taille! Pourquoi ne pas vous le dire tout de suite? OK, alors un sémaphore n'est pas pour vous. Bonne chance avec cette méthode!
    InformationsquelleAutor Vilx-
  9. -1

    Si vous souhaitez le maximum de débit, permettant à de multiples lecteurs à lire et à seulement un écrivain à écrire, la BCL a quelque chose qui s'appelle ReaderWriterLockSlim qui devrait l'aider à mincir votre code...

    InformationsquelleAutor DavidN