.NET에서 스레드 안전 차단 큐 구현

Question

.NET 용 스레드 안전 차단 큐의 구현을 찾고 있습니다. "스레드 안전 차단 대기열"이라는 의미 :--큐 메소드가 다른 스레드가 다른 스레드 (enqueue) 값을 넣을 때까지 스레드를 차단하는 큐에 대한 스레드 안전 액세스.

순간 나는 이것을 발견했다.http://www.eggeadcafe.com/articles/20060414.asp(그러나 .NET 1.1 용입니다).

누군가이 구현의 정확성을 언급/비판 할 수 있습니까? 또는 다른 것을 제안하십시오. 미리 감사드립니다.

Answer 1

이건 어때 .NET에서 차단 큐 생성?

.NET 1.1에 필요한 경우 (질문에서 확실하지 않은 경우) 제네릭을 떨어 뜨리고 교체하십시오. T ~와 함께 object.

Answer 2

참조를 위해 .NET 4를 소개합니다 System.Collections.Concurrent.BlockingCollection<T> 이 문제를 해결하기 위해 유형. 비 차단 큐의 경우 사용할 수 있습니다 System.Collections.Concurrent.ConcurrentQueue<T>. 주목하십시오 ConcurrentQueue<T> 아마도 기본 데이터 스토어로 사용될 것입니다. BlockingCollection<T> OP의 사용을 위해.

Answer 3

queue.synchronized http://msdn.microsoft.com/en-us/library/system.collections.queue.synchronized(vs.71).aspx

어쨌든 출발점입니다. 나는 차단 대기열을 사용한 적이 없습니다. 그렇게 관련이없는 게시물에 대해 죄송합니다.

Answer 4

Microsoft 예제는 좋은 예이지만 클래스에 캡슐화되지는 않습니다. 또한 소비자 스레드가 MTA에서 실행 중이라고 요구합니다 (WaitAny 호출로 인해). STA에서 실행 해야하는 경우가 있습니다 (예 : COM Interop을 수행하는 경우). 이 경우 Waitany를 사용할 수 없습니다.

이 문제를 극복하는 간단한 차단 큐 클래스가 있습니다.http://element533.blogspot.com/2010/01/stoppable-blocking-queue-for-net.html

Answer 5

예, .NET4에는 동시 컬렉션이 포함되어 있습니다. BTW, PFX 팀의 병렬 확장에 대한 매우 좋은 매뉴얼 - http://www.microsoft.com/downloads/details.aspx?familyid=86b3d32b-ad26-4bb8-a3ae-c1637026c3eee입니다.

PFX는 반응성 확장의 일부로 .NET 3.5에서도 사용할 수 있습니다.

Answer 6

Microsoft는 이것에 대한 꽤 좋은 샘플을 가지고 있습니다.

//Copyright (C) Microsoft Corporation.  All rights reserved.

using System;
using System.Threading;
using System.Collections;
using System.Collections.Generic;

// The thread synchronization events are encapsulated in this 
// class to allow them to easily be passed to the Consumer and 
// Producer classes. 
public class SyncEvents
{
    public SyncEvents()
    {
        // AutoResetEvent is used for the "new item" event because
        // we want this event to reset automatically each time the
        // consumer thread responds to this event.
        _newItemEvent = new AutoResetEvent(false);

        // ManualResetEvent is used for the "exit" event because
        // we want multiple threads to respond when this event is
        // signaled. If we used AutoResetEvent instead, the event
        // object would revert to a non-signaled state with after 
        // a single thread responded, and the other thread would 
        // fail to terminate.
        _exitThreadEvent = new ManualResetEvent(false);

        // The two events are placed in a WaitHandle array as well so
        // that the consumer thread can block on both events using
        // the WaitAny method.
        _eventArray = new WaitHandle[2];
        _eventArray[0] = _newItemEvent;
        _eventArray[1] = _exitThreadEvent;
    }

    // Public properties allow safe access to the events.
    public EventWaitHandle ExitThreadEvent
    {
        get { return _exitThreadEvent; }
    }
    public EventWaitHandle NewItemEvent
    {
        get { return _newItemEvent; }
    }
    public WaitHandle[] EventArray
    {
        get { return _eventArray; }
    }

    private EventWaitHandle _newItemEvent;
    private EventWaitHandle _exitThreadEvent;
    private WaitHandle[] _eventArray;
}

// The Producer class asynchronously (using a worker thread)
// adds items to the queue until there are 20 items.
public class Producer 
{
    public Producer(Queue<int> q, SyncEvents e)
    {
        _queue = q;
        _syncEvents = e;
    }
    public void ThreadRun()
    {
        int count = 0;
        Random r = new Random();
        while (!_syncEvents.ExitThreadEvent.WaitOne(0, false))
        {
            lock (((ICollection)_queue).SyncRoot)
            {
                while (_queue.Count < 20)
                {
                    _queue.Enqueue(r.Next(0, 100));
                    _syncEvents.NewItemEvent.Set();
                    count++;
                }
            }
        }
        Console.WriteLine("Producer thread: produced {0} items", count);
    }
    private Queue<int> _queue;
    private SyncEvents _syncEvents;
}

// The Consumer class uses its own worker thread to consume items
// in the queue. The Producer class notifies the Consumer class
// of new items with the NewItemEvent.
public class Consumer
{
    public Consumer(Queue<int> q, SyncEvents e)
    {
        _queue = q;
        _syncEvents = e;
    }
    public void ThreadRun()
    {
        int count = 0;
        while (WaitHandle.WaitAny(_syncEvents.EventArray) != 1)
        {
            lock (((ICollection)_queue).SyncRoot)
            {
                int item = _queue.Dequeue();
            }
            count++;
        }
        Console.WriteLine("Consumer Thread: consumed {0} items", count);
    }
    private Queue<int> _queue;
    private SyncEvents _syncEvents;
}

public class ThreadSyncSample
{
    private static void ShowQueueContents(Queue<int> q)
    {
        // Enumerating a collection is inherently not thread-safe,
        // so it is imperative that the collection be locked throughout
        // the enumeration to prevent the consumer and producer threads
        // from modifying the contents. (This method is called by the
        // primary thread only.)
        lock (((ICollection)q).SyncRoot)
        {
            foreach (int i in q)
            {
                Console.Write("{0} ", i);
            }
        }
        Console.WriteLine();
    }

    static void Main()
    {
        // Configure struct containing event information required
        // for thread synchronization. 
        SyncEvents syncEvents = new SyncEvents();

        // Generic Queue collection is used to store items to be 
        // produced and consumed. In this case 'int' is used.
        Queue<int> queue = new Queue<int>();

        // Create objects, one to produce items, and one to 
        // consume. The queue and the thread synchronization
        // events are passed to both objects.
        Console.WriteLine("Configuring worker threads...");
        Producer producer = new Producer(queue, syncEvents);
        Consumer consumer = new Consumer(queue, syncEvents);

        // Create the thread objects for producer and consumer
        // objects. This step does not create or launch the
        // actual threads.
        Thread producerThread = new Thread(producer.ThreadRun);
        Thread consumerThread = new Thread(consumer.ThreadRun);

        // Create and launch both threads.     
        Console.WriteLine("Launching producer and consumer threads...");        
        producerThread.Start();
        consumerThread.Start();

        // Let producer and consumer threads run for 10 seconds.
        // Use the primary thread (the thread executing this method)
        // to display the queue contents every 2.5 seconds.
        for (int i = 0; i < 4; i++)
        {
            Thread.Sleep(2500);
            ShowQueueContents(queue);
        }

        // Signal both consumer and producer thread to terminate.
        // Both threads will respond because ExitThreadEvent is a 
        // manual-reset event--so it stays 'set' unless explicitly reset.
        Console.WriteLine("Signaling threads to terminate...");
        syncEvents.ExitThreadEvent.Set();

        // Use Join to block primary thread, first until the producer thread
        // terminates, then until the consumer thread terminates.
        Console.WriteLine("main thread waiting for threads to finish...");
        producerThread.Join();
        consumerThread.Join();
    }
}

Answer 7

통화 코드를 잠그는 것이 완전히 제어 할 경우 더 나은 옵션 일 수 있습니다. 루프에서 대기열에 액세스하는 것을 고려하십시오. 잠금 잠금 장치를 여러 번 획득하여 성능 페널티가 발생할 수 있습니다.