Cola de tamaño fijo que quita automáticamente los valores antiguos en los nuevos enques

121

Estoy usando ConcurrentQueuepara una estructura de datos compartidos cuyo propósito es contener los últimos N objetos que se le pasaron (tipo de historial).

Supongamos que tenemos un navegador y queremos tener las últimas 100 URL examinadas. Quiero una cola que elimine automáticamente (retire) la entrada más antigua (primera) al insertar una nueva entrada (poner en cola) cuando se llene la capacidad (100 direcciones en el historial).

¿Cómo puedo lograr eso usando System.Collections?

Xaqron
fuente
No fue diseñado específicamente para usted, sino para cualquier persona que se encuentre con esta pregunta y pueda encontrarla útil. por cierto, también habla de C #. ¿Consiguió leer todas las respuestas (en 2 minutos) y se dio cuenta de que no hay código C # allí? De todos modos, yo mismo no estoy seguro, y por lo tanto es un comentario ...
Puede simplemente envolver los métodos en un candado. Dado que son rápidos, puede bloquear toda la matriz. Sin embargo, esto probablemente sea un engaño. La búsqueda de implementaciones de búfer circular con código C # puede encontrar algo. De todos modos, buena suerte.

Respuestas:

111

Escribiría una clase contenedora que en Enqueue verificaría el Count y luego Dequeue cuando el conteo exceda el límite.

 public class FixedSizedQueue<T>
 {
     ConcurrentQueue<T> q = new ConcurrentQueue<T>();
     private object lockObject = new object();

     public int Limit { get; set; }
     public void Enqueue(T obj)
     {
        q.Enqueue(obj);
        lock (lockObject)
        {
           T overflow;
           while (q.Count > Limit && q.TryDequeue(out overflow)) ;
        }
     }
 }
Richard Schneider
fuente
4
qes privado para el objeto, por lo que lockevitará que otros subprocesos tengan acceso simultáneo.
Richard Schneider
14
No es buena idea bloquear. El propósito de las colecciones concurrentes de BCL es proporcionar concurrencia sin bloqueo por motivos de rendimiento. El bloqueo de su código compromete ese beneficio. De hecho, no veo ninguna razón por la que necesites bloquear el deq.
KFL
2
@KFL, necesita bloquear porque County TryDequeueson dos operaciones independientes a las que no les importa sincronizar BCL Concurrent.
Richard Schneider
9
@RichardSchneider Si necesita manejar los problemas de concurrencia usted mismo, sería una buena idea cambiar el ConcurrentQueue<T>objeto por un Queue<T>objeto que sea más liviano.
0b101010
6
No defina su propia cola, solo use la heredada. Si hace lo que hace, en realidad no puede hacer nada más con los valores de la cola, todas las demás funciones, pero su nueva Enqueueaún llamará a la cola original. En otras palabras, aunque esta respuesta está marcada como aceptada, está completamente rota.
Gábor
104

Optaría por una ligera variante ... extender ConcurrentQueue para poder usar extensiones Linq en FixedSizeQueue

public class FixedSizedQueue<T> : ConcurrentQueue<T>
{
    private readonly object syncObject = new object();

    public int Size { get; private set; }

    public FixedSizedQueue(int size)
    {
        Size = size;
    }

    public new void Enqueue(T obj)
    {
        base.Enqueue(obj);
        lock (syncObject)
        {
            while (base.Count > Size)
            {
                T outObj;
                base.TryDequeue(out outObj);
            }
        }
    }
}
Dave Lawrence
fuente
1
qué sucede cuando alguien conoce estáticamente la instancia como ConcurrentQueue <T>, acaba de eludir su palabra clave "nueva".
mhand
6
@mhand Si 'alguien' quisiera hacer eso; entonces habrían elegido usar un objeto ConcurrentQueue <T> para empezar ... Esta es una clase de almacenamiento personalizada. Nadie está buscando que esto se envíe al marco .NET. Ha tratado de crear un problema por el simple hecho de hacerlo.
Dave Lawrence
9
mi punto es que, en lugar de subclasificar, tal vez debería simplemente envolver la cola ... esto aplica el comportamiento deseado en todos los casos. Además, dado que es una clase de almacenamiento personalizada, hagámosla completamente personalizada, solo exponga las operaciones que necesitamos, la subclasificación es la herramienta incorrecta aquí en mi humilde opinión.
mhand
3
@mhand Sí, entiendo lo que está diciendo ... Podría ajustar una cola y exponer el enumerador de la cola para hacer uso de las extensiones Linq.
Dave Lawrence
1
Estoy de acuerdo con @mhand en que no debes heredar ConcurrentQueue porque el método Enqueue no es virtual. Debe utilizar el proxy de la cola e implementar toda la interfaz si lo desea.
Chris Marisic
29

Para cualquiera que lo encuentre útil, aquí hay un código de trabajo basado en la respuesta de Richard Schneider anterior:

public class FixedSizedQueue<T>
{
    readonly ConcurrentQueue<T> queue = new ConcurrentQueue<T>();

    public int Size { get; private set; }

    public FixedSizedQueue(int size)
    {
        Size = size;
    }

    public void Enqueue(T obj)
    {
        queue.Enqueue(obj);

        while (queue.Count > Size)
        {
            T outObj;
            queue.TryDequeue(out outObj);
        }
    }
}
Tod Thomson
fuente
1
Votar en contra por las razones mencionadas (bloquear cuando se usa una ConcurrentQueue es malo) además de no implementar ninguna de las interfaces necesarias para que esta sea una verdadera colección.
Josh
11

Por lo que vale, aquí hay un búfer circular liviano con algunos métodos marcados para uso seguro e inseguro.

public class CircularBuffer<T> : IEnumerable<T>
{
    readonly int size;
    readonly object locker;

    int count;
    int head;
    int rear;
    T[] values;

    public CircularBuffer(int max)
    {
        this.size = max;
        locker = new object();
        count = 0;
        head = 0;
        rear = 0;
        values = new T[size];
    }

    static int Incr(int index, int size)
    {
        return (index + 1) % size;
    }

    private void UnsafeEnsureQueueNotEmpty()
    {
        if (count == 0)
            throw new Exception("Empty queue");
    }

    public int Size { get { return size; } }
    public object SyncRoot { get { return locker; } }

    #region Count

    public int Count { get { return UnsafeCount; } }
    public int SafeCount { get { lock (locker) { return UnsafeCount; } } }
    public int UnsafeCount { get { return count; } }

    #endregion

    #region Enqueue

    public void Enqueue(T obj)
    {
        UnsafeEnqueue(obj);
    }

    public void SafeEnqueue(T obj)
    {
        lock (locker) { UnsafeEnqueue(obj); }
    }

    public void UnsafeEnqueue(T obj)
    {
        values[rear] = obj;

        if (Count == Size)
            head = Incr(head, Size);
        rear = Incr(rear, Size);
        count = Math.Min(count + 1, Size);
    }

    #endregion

    #region Dequeue

    public T Dequeue()
    {
        return UnsafeDequeue();
    }

    public T SafeDequeue()
    {
        lock (locker) { return UnsafeDequeue(); }
    }

    public T UnsafeDequeue()
    {
        UnsafeEnsureQueueNotEmpty();

        T res = values[head];
        values[head] = default(T);
        head = Incr(head, Size);
        count--;

        return res;
    }

    #endregion

    #region Peek

    public T Peek()
    {
        return UnsafePeek();
    }

    public T SafePeek()
    {
        lock (locker) { return UnsafePeek(); }
    }

    public T UnsafePeek()
    {
        UnsafeEnsureQueueNotEmpty();

        return values[head];
    }

    #endregion


    #region GetEnumerator

    public IEnumerator<T> GetEnumerator()
    {
        return UnsafeGetEnumerator();
    }

    public IEnumerator<T> SafeGetEnumerator()
    {
        lock (locker)
        {
            List<T> res = new List<T>(count);
            var enumerator = UnsafeGetEnumerator();
            while (enumerator.MoveNext())
                res.Add(enumerator.Current);
            return res.GetEnumerator();
        }
    }

    public IEnumerator<T> UnsafeGetEnumerator()
    {
        int index = head;
        for (int i = 0; i < count; i++)
        {
            yield return values[index];
            index = Incr(index, size);
        }
    }

    System.Collections.IEnumerator System.Collections.IEnumerable.GetEnumerator()
    {
        return this.GetEnumerator();
    }

    #endregion
}

Me gusta usar la Foo()/SafeFoo()/UnsafeFoo()convención:

  • Foolos métodos llaman UnsafeFoopor defecto.
  • UnsafeFoo Los métodos modifican el estado libremente sin un bloqueo, solo deben llamar a otros métodos no seguros.
  • SafeFoolos métodos llaman a UnsafeFoométodos dentro de un candado.

Es un poco detallado, pero comete errores obvios, como llamar a métodos inseguros fuera de un bloqueo en un método que se supone que es seguro para subprocesos, más evidente.

Julieta
fuente
5

Aquí está mi opinión sobre la cola de tamaño fijo

Utiliza Cola normal, para evitar la sobrecarga de sincronización cuando Countse utiliza la propiedad ConcurrentQueue. También se implementa IReadOnlyCollectionpara que se puedan utilizar los métodos LINQ. El resto es muy similar a las otras respuestas aquí.

[Serializable]
[DebuggerDisplay("Count = {" + nameof(Count) + "}, Limit = {" + nameof(Limit) + "}")]
public class FixedSizedQueue<T> : IReadOnlyCollection<T>
{
    private readonly Queue<T> _queue = new Queue<T>();
    private readonly object _lock = new object();

    public int Count { get { lock (_lock) { return _queue.Count; } } }
    public int Limit { get; }

    public FixedSizedQueue(int limit)
    {
        if (limit < 1)
            throw new ArgumentOutOfRangeException(nameof(limit));

        Limit = limit;
    }

    public FixedSizedQueue(IEnumerable<T> collection)
    {
        if (collection is null || !collection.Any())
           throw new ArgumentException("Can not initialize the Queue with a null or empty collection", nameof(collection));

        _queue = new Queue<T>(collection);
        Limit = _queue.Count;
    }

    public void Enqueue(T obj)
    {
        lock (_lock)
        {
            _queue.Enqueue(obj);

            while (_queue.Count > Limit)
                _queue.Dequeue();
        }
    }

    public void Clear()
    {
        lock (_lock)
            _queue.Clear();
    }

    public IEnumerator<T> GetEnumerator()
    {
        lock (_lock)
            return new List<T>(_queue).GetEnumerator();
    }

    IEnumerator IEnumerable.GetEnumerator()
    {
        return GetEnumerator();
    }
}
Ali Zahid
fuente
3

Solo por diversión, aquí hay otra implementación que creo que aborda la mayoría de las preocupaciones de los comentaristas. En particular, la seguridad de subprocesos se logra sin bloqueo y la implementación está oculta por la clase de envoltura.

public class FixedSizeQueue<T> : IReadOnlyCollection<T>
{
  private ConcurrentQueue<T> _queue = new ConcurrentQueue<T>();
  private int _count;

  public int Limit { get; private set; }

  public FixedSizeQueue(int limit)
  {
    this.Limit = limit;
  }

  public void Enqueue(T obj)
  {
    _queue.Enqueue(obj);
    Interlocked.Increment(ref _count);

    // Calculate the number of items to be removed by this thread in a thread safe manner
    int currentCount;
    int finalCount;
    do
    {
      currentCount = _count;
      finalCount = Math.Min(currentCount, this.Limit);
    } while (currentCount != 
      Interlocked.CompareExchange(ref _count, finalCount, currentCount));

    T overflow;
    while (currentCount > finalCount && _queue.TryDequeue(out overflow))
      currentCount--;
  }

  public int Count
  {
    get { return _count; }
  }

  public IEnumerator<T> GetEnumerator()
  {
    return _queue.GetEnumerator();
  }

  System.Collections.IEnumerator System.Collections.IEnumerable.GetEnumerator()
  {
    return _queue.GetEnumerator();
  }
}
erdomke
fuente
1
Esto se rompe si se usa al mismo tiempo: ¿qué pasa si un hilo se reemplaza después de llamar _queue.Enqueue(obj)pero antes Interlocked.Increment(ref _count), y el otro hilo llama .Count? Obtendría un recuento incorrecto. No he comprobado los otros problemas.
KFL
3

Mi versión es solo una subclase de las normales Queue... nada especial, pero al ver a todos participando y todavía va con el título del tema, mejor podría ponerlo aquí. También devuelve los quitados de la cola por si acaso.

public sealed class SizedQueue<T> : Queue<T>
{
    public int FixedCapacity { get; }
    public SizedQueue(int fixedCapacity)
    {
        this.FixedCapacity = fixedCapacity;
    }

    /// <summary>
    /// If the total number of item exceed the capacity, the oldest ones automatically dequeues.
    /// </summary>
    /// <returns>The dequeued value, if any.</returns>
    public new T Enqueue(T item)
    {
        base.Enqueue(item);
        if (base.Count > FixedCapacity)
        {
            return base.Dequeue();
        }
        return default;
    }
}
5argón
fuente
2

Agreguemos una respuesta más. ¿Por qué esto sobre otros?

1) Sencillez. Tratar de garantizar el tamaño es bueno, pero conduce a una complejidad innecesaria que puede presentar sus propios problemas.

2) Implementa IReadOnlyCollection, lo que significa que puede usar Linq en él y pasarlo a una variedad de cosas que esperan IEnumerable.

3) Sin bloqueo. Muchas de las soluciones anteriores usan candados, lo cual es incorrecto en una colección sin candados.

4) Implementa el mismo conjunto de métodos, propiedades e interfaces que hace ConcurrentQueue, incluido IProducerConsumerCollection, que es importante si desea utilizar la colección con BlockingCollection.

Esta implementación podría potencialmente terminar con más entradas de las esperadas si TryDequeue falla, pero la frecuencia con la que esto ocurre no parece ser un código especializado que inevitablemente obstaculizará el rendimiento y causará sus propios problemas inesperados.

Si absolutamente desea garantizar un tamaño, implementar un Prune () o un método similar parece ser la mejor idea. Puede usar un bloqueo de lectura ReaderWriterLockSlim en los otros métodos (incluido TryDequeue) y tomar un bloqueo de escritura solo al podar.

class ConcurrentFixedSizeQueue<T> : IProducerConsumerCollection<T>, IReadOnlyCollection<T>, ICollection {
    readonly ConcurrentQueue<T> m_concurrentQueue;
    readonly int m_maxSize;

    public int Count => m_concurrentQueue.Count;
    public bool IsEmpty => m_concurrentQueue.IsEmpty;

    public ConcurrentFixedSizeQueue (int maxSize) : this(Array.Empty<T>(), maxSize) { }

    public ConcurrentFixedSizeQueue (IEnumerable<T> initialCollection, int maxSize) {
        if (initialCollection == null) {
            throw new ArgumentNullException(nameof(initialCollection));
        }

        m_concurrentQueue = new ConcurrentQueue<T>(initialCollection);
        m_maxSize = maxSize;
    }

    public void Enqueue (T item) {
        m_concurrentQueue.Enqueue(item);

        if (m_concurrentQueue.Count > m_maxSize) {
            T result;
            m_concurrentQueue.TryDequeue(out result);
        }
    }

    public void TryPeek (out T result) => m_concurrentQueue.TryPeek(out result);
    public bool TryDequeue (out T result) => m_concurrentQueue.TryDequeue(out result);

    public void CopyTo (T[] array, int index) => m_concurrentQueue.CopyTo(array, index);
    public T[] ToArray () => m_concurrentQueue.ToArray();

    public IEnumerator<T> GetEnumerator () => m_concurrentQueue.GetEnumerator();
    IEnumerator IEnumerable.GetEnumerator () => GetEnumerator();

    // Explicit ICollection implementations.
    void ICollection.CopyTo (Array array, int index) => ((ICollection)m_concurrentQueue).CopyTo(array, index);
    object ICollection.SyncRoot => ((ICollection) m_concurrentQueue).SyncRoot;
    bool ICollection.IsSynchronized => ((ICollection) m_concurrentQueue).IsSynchronized;

    // Explicit IProducerConsumerCollection<T> implementations.
    bool IProducerConsumerCollection<T>.TryAdd (T item) => ((IProducerConsumerCollection<T>) m_concurrentQueue).TryAdd(item);
    bool IProducerConsumerCollection<T>.TryTake (out T item) => ((IProducerConsumerCollection<T>) m_concurrentQueue).TryTake(out item);

    public override int GetHashCode () => m_concurrentQueue.GetHashCode();
    public override bool Equals (object obj) => m_concurrentQueue.Equals(obj);
    public override string ToString () => m_concurrentQueue.ToString();
}
Josh
fuente
2

Solo porque nadie lo ha dicho todavía ... puedes usar a LinkedList<T>y agregar la seguridad de subprocesos:

public class Buffer<T> : LinkedList<T>
{
    private int capacity;

    public Buffer(int capacity)
    {
        this.capacity = capacity;   
    }

    public void Enqueue(T item)
    {
        // todo: add synchronization mechanism
        if (Count == capacity) RemoveLast();
        AddFirst(item);
    }

    public T Dequeue()
    {
        // todo: add synchronization mechanism
        var last = Last.Value;
        RemoveLast();
        return last;
    }
}

Una cosa a tener en cuenta es que el orden de enumeración predeterminado será LIFO en este ejemplo. Pero eso puede anularse si es necesario.

Brandon
fuente
1

Para su placer de codificación, le presento el ' ConcurrentDeck'

public class ConcurrentDeck<T>
{
   private readonly int _size;
   private readonly T[] _buffer;
   private int _position = 0;

   public ConcurrentDeck(int size)
   {
       _size = size;
       _buffer = new T[size];
   }

   public void Push(T item)
   {
       lock (this)
       {
           _buffer[_position] = item;
           _position++;
           if (_position == _size) _position = 0;
       }
   }

   public T[] ReadDeck()
   {
       lock (this)
       {
           return _buffer.Skip(_position).Union(_buffer.Take(_position)).ToArray();
       }
   }
}

Ejemplo de uso:

void Main()
{
    var deck = new ConcurrentDeck<Tuple<string,DateTime>>(25);
    var handle = new ManualResetEventSlim();
    var task1 = Task.Factory.StartNew(()=>{
    var timer = new System.Timers.Timer();
    timer.Elapsed += (s,a) => {deck.Push(new Tuple<string,DateTime>("task1",DateTime.Now));};
    timer.Interval = System.TimeSpan.FromSeconds(1).TotalMilliseconds;
    timer.Enabled = true;
    handle.Wait();
    }); 
    var task2 = Task.Factory.StartNew(()=>{
    var timer = new System.Timers.Timer();
    timer.Elapsed += (s,a) => {deck.Push(new Tuple<string,DateTime>("task2",DateTime.Now));};
    timer.Interval = System.TimeSpan.FromSeconds(.5).TotalMilliseconds;
    timer.Enabled = true;
    handle.Wait();
    }); 
    var task3 = Task.Factory.StartNew(()=>{
    var timer = new System.Timers.Timer();
    timer.Elapsed += (s,a) => {deck.Push(new Tuple<string,DateTime>("task3",DateTime.Now));};
    timer.Interval = System.TimeSpan.FromSeconds(.25).TotalMilliseconds;
    timer.Enabled = true;
    handle.Wait();
    }); 
    System.Threading.Thread.Sleep(TimeSpan.FromSeconds(10));
    handle.Set();
    var outputtime = DateTime.Now;
    deck.ReadDeck().Select(d => new {Message = d.Item1, MilliDiff = (outputtime - d.Item2).TotalMilliseconds}).Dump(true);
}
Chris Hayes
fuente
1
Me gusta esta implementación, pero tenga en cuenta que cuando no se ha agregado ninguna, devuelve el valor predeterminado (T)
Daniel Leach
Si usa el bloqueo de esta manera, debe usar ReaderWriterLockSlim para priorizar a sus lectores.
Josh
1

Bueno, depende del uso, he notado que algunas de las soluciones anteriores pueden exceder el tamaño cuando se usan en un entorno de subprocesos múltiples. De todos modos, mi caso de uso fue mostrar los últimos 5 eventos y hay varios subprocesos que escriben eventos en la cola y otro subproceso que lee y lo muestra en un control de Winform. Entonces esta fue mi solución.

EDITAR: Dado que ya estamos usando el bloqueo dentro de nuestra implementación, realmente no necesitamos ConcurrentQueue, puede mejorar el rendimiento.

class FixedSizedConcurrentQueue<T> 
{
    readonly Queue<T> queue = new Queue<T>();
    readonly object syncObject = new object();

    public int MaxSize { get; private set; }

    public FixedSizedConcurrentQueue(int maxSize)
    {
        MaxSize = maxSize;
    }

    public void Enqueue(T obj)
    {
        lock (syncObject)
        {
            queue.Enqueue(obj);
            while (queue.Count > MaxSize)
            {
                queue.Dequeue();
            }
        }
    }

    public T[] ToArray()
    {
        T[] result = null;
        lock (syncObject)
        {
            result = queue.ToArray();
        }

        return result;
    }

    public void Clear()
    {
        lock (syncObject)
        {
            queue.Clear();
        }
    }
}

EDITAR: Realmente no necesitamos syncObjecten el ejemplo anterior y podemos usar el queueobjeto ya que no estamos reinicializando queueen ninguna función y está marcado como de readonlytodos modos.

Mubashar
fuente
0

La respuesta aceptada tendrá efectos secundarios evitables.

Mecanismos de bloqueo y sin bloqueo de grano fino

Los enlaces a continuación son referencias que utilicé cuando escribí mi ejemplo a continuación.

Si bien la documentación de Microsoft es un poco engañosa, ya que usan un bloqueo, sin embargo, bloquean las clases de segmento. Las propias clases de segmento utilizan Interlocked.

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;

namespace Lib.Core
{
    // Sources: 
    // https://docs.microsoft.com/en-us/dotnet/standard/collections/thread-safe/
    // https://docs.microsoft.com/en-us/dotnet/api/system.threading.interlocked?view=netcore-3.1
    // https://github.com/dotnet/runtime/blob/master/src/libraries/System.Private.CoreLib/src/System/Collections/Concurrent/ConcurrentQueue.cs
    // https://github.com/dotnet/runtime/blob/master/src/libraries/System.Private.CoreLib/src/System/Collections/Concurrent/ConcurrentQueueSegment.cs

    /// <summary>
    /// Concurrent safe circular buffer that will used a fixed capacity specified and resuse slots as it goes.
    /// </summary>
    /// <typeparam name="TObject">The object that you want to go into the slots.</typeparam>
    public class ConcurrentCircularBuffer<TObject>
    {
        private readonly ConcurrentQueue<TObject> _queue;

        public int Capacity { get; private set; }

        public ConcurrentCircularBuffer(int capacity)
        {
            if(capacity <= 0)
            {
                throw new ArgumentException($"The capacity specified '{capacity}' is not valid.", nameof(capacity));
            }

            // Setup the queue to the initial capacity using List's underlying implementation.
            _queue = new ConcurrentQueue<TObject>(new List<TObject>(capacity));

            Capacity = capacity;
        }

        public void Enqueue(TObject @object)
        {
            // Enforce the capacity first so the head can be used instead of the entire segment (slow).
            while (_queue.Count + 1 > Capacity)
            {
                if (!_queue.TryDequeue(out _))
                {
                    // Handle error condition however you want to ie throw, return validation object, etc.
                    var ex = new Exception("Concurrent Dequeue operation failed.");
                    ex.Data.Add("EnqueueObject", @object);
                    throw ex;
                }
            }

            // Place the item into the queue
            _queue.Enqueue(@object);
        }

        public TObject Dequeue()
        {
            if(_queue.TryDequeue(out var result))
            {
                return result;
            }

            return default;
        }
    }
}
jjhayter
fuente
0

Aquí hay otra implementación que usa el ConcurrentQueue subyacente tanto como sea posible mientras proporciona las mismas interfaces disponibles a través de ConcurrentQueue.

/// <summary>
/// This is a FIFO concurrent queue that will remove the oldest added items when a given limit is reached.
/// </summary>
/// <typeparam name="TValue"></typeparam>
public class FixedSizedConcurrentQueue<TValue> : IProducerConsumerCollection<TValue>, IReadOnlyCollection<TValue>
{
    private readonly ConcurrentQueue<TValue> _queue;

    private readonly object _syncObject = new object();

    public int LimitSize { get; }

    public FixedSizedConcurrentQueue(int limit)
    {
        _queue = new ConcurrentQueue<TValue>();
        LimitSize = limit;
    }

    public FixedSizedConcurrentQueue(int limit, System.Collections.Generic.IEnumerable<TValue> collection)
    {
        _queue = new ConcurrentQueue<TValue>(collection);
        LimitSize = limit;

    }

    public int Count => _queue.Count;

    bool ICollection.IsSynchronized => ((ICollection) _queue).IsSynchronized;

    object ICollection.SyncRoot => ((ICollection)_queue).SyncRoot; 

    public bool IsEmpty => _queue.IsEmpty;

    // Not supported until .NET Standard 2.1
    //public void Clear() => _queue.Clear();

    public void CopyTo(TValue[] array, int index) => _queue.CopyTo(array, index);

    void ICollection.CopyTo(Array array, int index) => ((ICollection)_queue).CopyTo(array, index);

    public void Enqueue(TValue obj)
    {
        _queue.Enqueue(obj);
        lock( _syncObject )
        {
            while( _queue.Count > LimitSize ) {
                _queue.TryDequeue(out _);
            }
        }
    }

    public IEnumerator<TValue> GetEnumerator() => _queue.GetEnumerator();

    IEnumerator IEnumerable.GetEnumerator() => ((IEnumerable<TValue>)this).GetEnumerator();

    public TValue[] ToArray() => _queue.ToArray();

    public bool TryAdd(TValue item)
    {
        Enqueue(item);
        return true;
    }

    bool IProducerConsumerCollection<TValue>.TryTake(out TValue item) => TryDequeue(out item);

    public bool TryDequeue(out TValue result) => _queue.TryDequeue(out result);

    public bool TryPeek(out TValue result) => _queue.TryPeek(out result);

}
Tod Cunningham
fuente
-1

Esta es mi versión de la cola:

public class FixedSizedQueue<T> {
  private object LOCK = new object();
  ConcurrentQueue<T> queue;

  public int MaxSize { get; set; }

  public FixedSizedQueue(int maxSize, IEnumerable<T> items = null) {
     this.MaxSize = maxSize;
     if (items == null) {
        queue = new ConcurrentQueue<T>();
     }
     else {
        queue = new ConcurrentQueue<T>(items);
        EnsureLimitConstraint();
     }
  }

  public void Enqueue(T obj) {
     queue.Enqueue(obj);
     EnsureLimitConstraint();
  }

  private void EnsureLimitConstraint() {
     if (queue.Count > MaxSize) {
        lock (LOCK) {
           T overflow;
           while (queue.Count > MaxSize) {
              queue.TryDequeue(out overflow);
           }
        }
     }
  }


  /// <summary>
  /// returns the current snapshot of the queue
  /// </summary>
  /// <returns></returns>
  public T[] GetSnapshot() {
     return queue.ToArray();
  }
}

Encuentro útil tener un constructor que se basa en un IEnumerable y me parece útil tener un GetSnapshot para tener una lista segura de múltiples subprocesos (matriz en este caso) de los elementos en el momento de la llamada, que no aumenta errores si cambia la colección subyacente.

La verificación de la cuenta doble es para evitar el bloqueo en algunas circunstancias.

No importante
fuente
1
Votar en contra para bloquear la cola. Si absolutamente desea bloquear, un ReaderWriterLockSlim sería mejor (asumiendo que espera tomar un bloqueo de lectura con más frecuencia que un bloqueo de escritura). GetSnapshot tampoco es necesario. Si implementa IReadOnlyCollection <T> (que debería para la semántica de IEnumerable), ToList () tendrá la misma función.
Josh
ConcurrentQueue maneja los bloqueos en su implementación, vea los enlaces en mi respuesta.
jjhayter