Membuat Antrian pemblokiran <T> di .NET?

Saya memiliki skenario di mana saya memiliki beberapa utas yang menambah antrian dan beberapa utas membaca dari antrian yang sama. Jika antrian mencapai ukuran tertentu semua utas yang mengisi antrian akan diblokir saat ditambahkan hingga item dihapus dari antrian.

Solusi di bawah ini adalah apa yang saya gunakan saat ini dan pertanyaan saya adalah: Bagaimana ini bisa diperbaiki? Apakah ada objek yang sudah mengaktifkan perilaku ini di BCL yang seharusnya saya gunakan?

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

Itu terlihat sangat tidak aman (sinkronisasi sangat sedikit); bagaimana dengan sesuatu seperti:

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)

Pada kenyataannya, Anda ingin cara untuk menutup antrian sehingga pembaca mulai keluar dengan bersih - mungkin sesuatu seperti bool flag - jika diatur, antrian kosong hanya kembali (daripada memblokir):

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

Gunakan .net 4 BlockingCollection, untuk enqueue gunakan Add (), untuk dequeue gunakan Take (). Ini secara internal menggunakan ConcurrentQueue non-blocking. Info lebih lanjut di sini Cepat dan Terbaik Produser / teknik antrian konsumen BlockingCollection vs Antrian bersamaan

"Bagaimana ini bisa diperbaiki?"

Nah, Anda perlu melihat setiap metode di kelas Anda dan mempertimbangkan apa yang akan terjadi jika utas lain secara bersamaan memanggil metode itu atau metode lainnya. Misalnya, Anda meletakkan kunci di metode Hapus, tetapi tidak di metode Tambahkan. Apa yang terjadi jika satu utas Menambahkan pada saat yang sama dengan utas lainnya Menghapus?Hal buruk.

Juga pertimbangkan bahwa metode dapat mengembalikan objek kedua yang menyediakan akses ke data internal objek pertama - misalnya, GetEnumerator. Bayangkan satu utas melewati enumerator itu, utas lain memodifikasi daftar pada saat yang sama.Tidak baik.

Aturan praktis yang baik adalah membuat ini lebih sederhana untuk mendapatkan yang benar dengan mengurangi jumlah metode di kelas ke minimum absolut.

Secara khusus, jangan mewarisi kelas wadah lain, karena Anda akan mengekspos semua metode kelas itu, menyediakan cara bagi pemanggil untuk merusak data internal, atau untuk melihat sebagian menyelesaikan perubahan pada data (sama buruknya, karena data muncul rusak pada saat itu). Sembunyikan semua detail dan benar-benar kejam tentang bagaimana Anda mengizinkannya.

Saya sangat menyarankan Anda untuk menggunakan solusi di luar rak - dapatkan buku tentang threading atau gunakan perpustakaan pihak ke-3. Kalau tidak, mengingat apa yang Anda coba, Anda akan men-debug kode Anda untuk waktu yang lama.

Juga, bukankah lebih masuk akal bagi Hapus untuk mengembalikan item (katakanlah, item yang ditambahkan pertama kali, karena itu adalah antrian), daripada penelepon memilih item tertentu? Dan ketika antrian kosong, mungkin Hapus juga harus diblokir.

Pembaruan: Jawaban Marc sebenarnya mengimplementasikan semua saran ini! :) Tapi saya akan meninggalkan ini di sini karena mungkin membantu untuk memahami mengapa versinya adalah perbaikan.

Anda bisa menggunakan BlockingCollection dan ConcurrentQueue di System.Collections.Concurrent Namespace

 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)
    {
    }



}

Saya baru saja mengetuk ini menggunakan Ekstensi Reaktif dan ingat pertanyaan ini:

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

Tidak harus sepenuhnya aman, tetapi sangat sederhana.

Inilah yang saya datang op untuk antrian memblokir aman dibatasi thread.

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

Saya belum sepenuhnya menjelajahi TPL tetapi mereka mungkin memiliki sesuatu yang sesuai dengan kebutuhan Anda, atau setidaknya, beberapa reflektor makanan ternak untuk mengambil inspirasi dari.

Semoga itu bisa membantu.

Nah, Anda mungkin melihat System.Threading.Semaphorekelas. Selain itu - tidak, Anda harus membuatnya sendiri. AFAIK tidak ada koleksi bawaan seperti itu.

Jika Anda ingin throughput maksimum, memungkinkan banyak pembaca untuk membaca dan hanya satu penulis untuk menulis, BCL memiliki sesuatu yang disebut ReaderWriterLockSlim yang seharusnya membantu menurunkan kode Anda ...