Создание блокирующейся Очереди <T> в.NET?

"Как это может быть улучшено?"

ну, необходимо посмотреть на каждый метод в классе и рассмотреть то, что произошло бы, если бы другой поток одновременно называл тот метод или какой-либо другой метод. Например, Вы вставляете блокировку Удалить метод, но не в Добавить методе. Что происходит, если один поток Добавляет в то же время, что и другой поток Удаляет? Плохие вещи.

Также полагают, что метод может возвратить второй объект, который обеспечивает доступ к внутренним данным первого объекта - например, GetEnumerator. Предположите, что один поток проходит тот перечислитель, другой поток изменяет список одновременно. Не хороший.

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

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

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

кроме того, разве это не имело бы большего смысла для, Удаляют для возврата, объект (скажите, тот, который был добавлен сначала, поскольку это - очередь), а не вызывающая сторона, выбирающая определенный объект? И то, когда очередь пуста, возможно, Удалите, должно также заблокироваться.

Обновление: ответ Marc на самом деле реализует все эти предложения!:) Но я оставлю это здесь, поскольку может быть полезно понять, почему его версия является таким улучшением.

Ну, Вы могли бы посмотреть System.Threading.Semaphore класс. Кроме этого - нет, необходимо сделать это сами. AFAIK там не является таким встроенным набором.

Я не полностью исследовал TPL, но у них могло бы быть что-то, что соответствует Вашим потребностям, или по крайней мере, некоторый фураж Отражателя для ловли некоторого вдохновения от.

Hope, которая помогает.

Если Вы хотите максимальную пропускную способность, позволяя нескольким читателям читать и только одно устройство записи для записи, BCL имеет что-то названное ReaderWriterLockSlim, который должен помочь сократить код...

Это то, что я сделал для поточно-безопасной ограниченной блокировки очередь.

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

Я только что сделал это с помощью Reactive Extensions и вспомнил этот вопрос:

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

Не обязательно полностью безопасно, но очень просто.