Очередь фиксированного размера, которая автоматически удаляет старые значения из очереди при новых запросах

121

Я использую ConcurrentQueueобщую структуру данных, цель которой - хранить последние N переданных ей объектов (своего рода история).

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

Как я могу этого добиться System.Collections?

Xaqron
источник
Он не предназначен специально для вас, но предназначен для всех, кто сталкивается с этим вопросом и может найти его полезным. Кстати, он тоже говорит о C #. Удалось ли вам прочитать все ответы (за 2 минуты) и выяснить, что там нет кода C #? Во всяком случае, я сам не уверен, и, следовательно, это комментарий ...
Вы можете просто заключить методы в блокировку. Учитывая, что они быстрые, вы можете просто заблокировать весь массив. Хотя, наверное, это обман. Поиск реализаций кольцевого буфера с помощью кода C # может кое-что вам найти. В любом случае, удачи.

Ответы:

111

Я бы написал класс-оболочку, который в Enqueue будет проверять счетчик, а затем выводить из очереди, когда счет превышает предел.

 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)) ;
        }
     }
 }
Ричард Шнайдер
источник
4
qявляется закрытым для объекта, поэтому он lockне позволит другим потокам одновременный доступ.
Ричард Шнайдер
14
Блокировать - не лучшая идея. Вся цель параллельных коллекций BCL - обеспечить параллелизм без блокировок по соображениям производительности. Блокировка в вашем коде ставит под угрозу это преимущество. На самом деле я не вижу причин, по которым вам нужно заблокировать deq.
KFL
2
@KFL, нужно заблокировать , потому что Countи TryDequeueдве независимые операции, уход не синхронизируется по BCL Concurrent.
Ричард Шнайдер
9
@RichardSchneider Если вам нужно самостоятельно решать проблемы с параллелизмом, было бы неплохо поменять ConcurrentQueue<T>объект на Queue<T>более легкий.
0b101010 04
6
Не определяйте собственную очередь, просто используйте унаследованную. Если вы сделаете то же самое, вы больше ничего не сможете сделать со значениями очереди, все другие функции, кроме вашей новой Enqueue, по-прежнему будут вызывать исходную очередь. Другими словами, хотя этот ответ отмечен как принятый, он полностью неверен.
Gábor
104

Я бы выбрал небольшой вариант ... расширил ConcurrentQueue, чтобы иметь возможность использовать расширения Linq в 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);
            }
        }
    }
}
Дэйв Лоуренс
источник
1
что происходит, когда кто-то статически знает экземпляр как ConcurrentQueue <T>, он только что обошел ваше ключевое слово 'new'.
mhand 03
6
@mhand Если "кто-то" хотел это сделать; тогда они предпочли бы использовать объект ConcurrentQueue <T> для начала ... Это настраиваемый класс хранения. Никто не хочет, чтобы это было внесено в платформу .NET. Вы пытались создать проблему ради этого.
Дэйв Лоуренс
9
Моя точка зрения заключается в том, что вместо создания подклассов вам следует просто обернуть очередь ... это обеспечивает желаемое поведение во всех случаях. Кроме того, поскольку это настраиваемый класс хранилища, давайте сделаем его полностью настраиваемым, выставим только те операции, которые нам нужны, создание подклассов - неправильный инструмент здесь ИМХО.
mhand 04
3
@mhand Да, я понимаю, о чем вы говорите ... Я мог бы обернуть очередь и открыть перечислитель очереди, чтобы использовать расширения Linq.
Дэйв Лоуренс
1
Я согласен с @mhand, вы не должны наследовать ConcurrentQueue, потому что метод Enqueue не виртуальный. Вы должны проксировать очередь и при желании реализовать весь интерфейс.
Крис Марисич
29

Для тех, кто считает это полезным, вот рабочий код, основанный на ответе Ричарда Шнайдера выше:

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);
        }
    }
}
Тод Томсон
источник
1
Голосование по указанным причинам (блокировка при использовании ConcurrentQueue - это плохо) в дополнение к тому, что не реализованы какие-либо необходимые интерфейсы для того, чтобы это была настоящая коллекция.
Джош
11

Как бы то ни было, вот легкий кольцевой буфер с некоторыми методами, отмеченными для безопасного и небезопасного использования.

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
}

Мне нравится использовать Foo()/SafeFoo()/UnsafeFoo()соглашение:

  • Fooметоды вызывают UnsafeFooпо умолчанию.
  • UnsafeFoo методы изменяют состояние свободно без блокировки, они должны вызывать только другие небезопасные методы.
  • SafeFooметоды вызывают UnsafeFooметоды внутри блокировки.

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

Джульетта
источник
5

Вот мой взгляд на очередь фиксированного размера

Он использует обычную очередь, чтобы избежать накладных расходов на синхронизацию при использовании Countсвойства ConcurrentQueue. Он также реализует IReadOnlyCollectionтак, чтобы можно было использовать методы LINQ. Остальное очень похоже на другие ответы здесь.

[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();
    }
}
Али Захид
источник
3

Ради удовольствия, вот еще одна реализация, которая, как мне кажется, решает большинство проблем комментаторов. В частности, потокобезопасность достигается без блокировки, а реализация скрыта классом упаковки.

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
источник
1
Это нарушается при одновременном использовании - что, если поток прерывается после вызова, _queue.Enqueue(obj)но до этого Interlocked.Increment(ref _count), а другой поток вызывает .Count? Это будет неправильный счет. Я не проверял другие проблемы.
KFL
3

Моя версия - это просто подкласс обычных Queue... ничего особенного, кроме того, что я вижу, что все участвуют, и она по-прежнему соответствует названию темы, я мог бы также поместить ее здесь. Он также на всякий случай возвращает исключенные из очереди.

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;
    }
}
5argon
источник
2

Добавим еще один ответ. Почему это по сравнению с другими?

1) Простота. Попытка гарантировать размер - это хорошо, но приводит к ненужным сложностям, которые могут иметь свои собственные проблемы.

2) Реализует IReadOnlyCollection, то есть вы можете использовать на нем Linq и передавать его во множество вещей, ожидающих IEnumerable.

3) Без блокировки. Многие из вышеперечисленных решений используют блокировки, что неверно для коллекции без блокировки.

4) Реализует тот же набор методов, свойств и интерфейсов, что и ConcurrentQueue, включая IProducerConsumerCollection, что важно, если вы хотите использовать коллекцию с BlockingCollection.

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

Если вы абсолютно хотите гарантировать размер, реализация Prune () или аналогичного метода кажется лучшей идеей. Вы можете использовать блокировку чтения ReaderWriterLockSlim в других методах (включая TryDequeue) и использовать блокировку записи только при сокращении.

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();
}
мистифицировать
источник
2

Просто потому, что этого еще никто не сказал ... вы можете использовать a LinkedList<T>и добавить потокобезопасность:

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

Следует отметить, что в этом примере порядок перечисления по умолчанию будет LIFO. Но при необходимости это можно изменить.

Brandon
источник
1

Для вашего удовольствия от программирования я представляю вам ' 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();
       }
   }
}

Пример использования:

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);
}
Крис Хейс
источник
1
Мне нравится эта реализация, но обратите внимание, что когда ничего не было добавлено, она возвращает значение по умолчанию (T)
Дэниел Лич
Если вы используете блокировку таким образом, вы должны использовать ReaderWriterLockSlim для определения приоритетов ваших читателей.
Джош
1

Что ж, это зависит от использования. Я заметил, что некоторые из вышеперечисленных решений могут превышать размер при использовании в многопоточной среде. В любом случае мой вариант использования заключался в отображении последних 5 событий, и есть несколько потоков, записывающих события в очередь, и еще один поток, читающий из нее и отображающий его в элементе управления Winform. Итак, это было моим решением.

EDIT: поскольку мы уже используем блокировку в нашей реализации, нам действительно не нужен ConcurrentQueue, это может улучшить производительность.

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

РЕДАКТИРОВАТЬ: Нам действительно не нужен syncObjectприведенный выше пример, и мы скорее можем использовать queueобъект, поскольку мы не повторно инициализируем queueкакую-либо функцию и readonlyвсе равно помечены как .

Mubashar
источник
0

Принятый ответ будет иметь побочные эффекты, которых можно избежать.

Мелкозернистые запорные и безблокирующие механизмы

Ссылки ниже - это ссылки, которые я использовал, когда писал свой пример ниже.

Хотя документация от Microsoft немного вводит в заблуждение, поскольку они используют блокировку, они, тем не менее, блокируют классы сегментов. Сами классы сегментов используют 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
источник
0

Вот еще одна реализация, которая максимально использует базовый ConcurrentQueue, обеспечивая при этом те же интерфейсы, которые доступны через 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);

}
Тод Каннингем
источник
-1

Это моя версия очереди:

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

Я считаю полезным иметь конструктор, построенный на IEnumerable, и я считаю полезным иметь GetSnapshot, чтобы иметь многопоточный безопасный список (в данном случае массив) элементов в момент вызова, который не поднимается ошибки при изменении подстилающей коллекции.

Проверка двойного счета предназначена для предотвращения блокировки в некоторых случаях.

Не важный
источник
1
Голосование за блокировку очереди. Если вы абсолютно хотите заблокировать, лучше всего подойдет ReaderWriterLockSlim (при условии, что вы ожидаете использовать блокировку чтения чаще, чем блокировку записи). GetSnapshot также не нужен. Если вы реализуете IReadOnlyCollection <T> (что необходимо для семантики IEnumerable), ToList () будет выполнять ту же функцию.
Джош
ConcurrentQueue обрабатывает блокировки в своей реализации, см. Ссылки в моем ответе.
jjhayter