Простая коллекция производитель-потребитель без BlockingCollection

Question

Я хочу написать простую очередь производитель-потребитель без использования встроенной System.Collections.Concurrent.BlockingCollection.Вот быстрая попытка, которая «кажется» работает.Что-то не так с этим в отношении потоков, условий гонки, взаимоблокировок и т. Д.?

class ProducerConsumerQueue<T>
{
    Queue<T> Queue = new Queue<T>();
    ManualResetEvent Event = new ManualResetEvent(false);
    object Lock = new object();

    public void Add(T t)
    {
        lock (Lock)
        {
            Queue.Enqueue(t);
        }
        Event.Set();
    }

    public bool TryTake(out T t, int timeout)
    {
        if (Event.WaitOne(timeout))
        {
            lock (Lock)
            {
                if (Queue.Count > 0)
                {
                    t = Queue.Dequeue();
                    if (Queue.Count == 0) Event.Reset();
                    return true;
                }
            }
        }
        t = default(T);
        return false;
    }
}

Кстати.мне нужны только два метода Add и TryTake, мне не нужно IEnumerable и т. д.

Answer 1

Согласно моему комментарию в вопросе,

Вот мое предлагаемое решение.

public class BlockingQueue<T>
{
    // In order to get rid of Lock object
    // Any thread should be able to add items to the queue
    private readonly ConcurrentQueue<T> _queue = new ConcurrentQueue<T>();

    // Only one thread is able to consume from queue
    // You can fine tune this to your interest
    private readonly SemaphoreSlim _slim = new SemaphoreSlim(1,1);

    public void Add(T item) {
        _queue.Enqueue(item);
    }

    public bool TryTake(out T item, TimeSpan timeout) {
        if (_slim.Wait(timeout)){
            return _queue.TryDequeue(out item);
        }
        item = default(T);
        return false;
    }
}

Answer 2

Microsoft недавно отказалась от System.Threading.Channels, который предназначен для предоставления оптимизированных API производителей / потребителей, что может хорошо подойти в этом случае.Он охватывает неограниченные и ограниченные сценарии и включает в себя сценарии «один плюс несколько читателей / писателей».API довольно прост и интуитивно понятен в использовании;Единственное небольшое предостережение заключается в том, что он использует async -ориентированный API (для потребителя и - в случае ограниченных каналов - для производителя).

Дело здесь в том, что код, который вы не 'Как правило, писать - это код, в котором меньше болевых точек, особенно если он написан командой, обладающей опытом и заинтересованностью в решении конкретных проблем.

---

Однако: вы можете делать все в своем текущем кодебез необходимости ManualResetEvent - lock в C # - это просто оболочка для простейших частей Monitor, но Monitor также обеспечивает функцию ожидания / импульса:

class ProducerConsumerQueue<T>
{
    private readonly Queue<T> Queue = new Queue<T>();

    public void Add(T t)
    {
        lock (Queue)
        {
            Queue.Enqueue(t);
            if (Queue.Count == 1)
            {
                // wake up one sleeper
                Monitor.Pulse(Queue);
            }
        }
    }

    public bool TryTake(out T t, int millisecondsTimeout)
    {
        lock (Queue)
        {
            if (Queue.Count == 0)
            {
                // try and wait for arrival
                Monitor.Wait(Queue, millisecondsTimeout);
            }
            if (Queue.Count != 0)
            {
                t = Queue.Dequeue();
                return true;
            }
        }
        t = default(T);
        return false;
    }
}

Answer 3

Я думаю, что использование lock и ManualResetEvent является избыточным.Я предлагаю вам прочитать больше о ManualResetEvent о том, как вводить и выходить из синхронизированных областей в вашем коде (вы также можете взглянуть на другие механизмы синхронизации, доступные в System.Threading ).

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

Надеюсь, это поможет!