C# producer/consumer

Question

저는 최근에 생산자/소비자 패턴 C# 구현을 접했습니다. 그것은 매우 간단합니다 (적어도 나를 위해) 매우 우아합니다.

가 2006 주위에 고안 한 것 같다, 그래서이 구현은
경우 내가 궁금

- 안전
- 여전히 적용

코드 (원본 코드가 http://bytes.com/topic/net/answers/575276-producer-consumer#post2251375에서 참조 된) 이하

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

public class Test 
{ 
    static ProducerConsumer queue; 

    static void Main() 
    { 
     queue = new ProducerConsumer(); 
     new Thread(new ThreadStart(ConsumerJob)).Start(); 

     Random rng = new Random(0); 
     for (int i=0; i < 10; i++) 
     { 
      Console.WriteLine ("Producing {0}", i); 
      queue.Produce(i); 
      Thread.Sleep(rng.Next(1000)); 
     } 
    } 

    static void ConsumerJob() 
    { 
     // Make sure we get a different random seed from the 
     // first thread 
     Random rng = new Random(1); 
     // We happen to know we've only got 10 
     // items to receive 
     for (int i=0; i < 10; i++) 
     { 
      object o = queue.Consume(); 
      Console.WriteLine ("\t\t\t\tConsuming {0}", o); 
      Thread.Sleep(rng.Next(1000)); 
     } 
    } 
} 

public class ProducerConsumer 
{ 
    readonly object listLock = new object(); 
    Queue queue = new Queue(); 

    public void Produce(object o) 
    { 
     lock (listLock) 
     { 
      queue.Enqueue(o); 

      // We always need to pulse, even if the queue wasn't 
      // empty before. Otherwise, if we add several items 
      // in quick succession, we may only pulse once, waking 
      // a single thread up, even if there are multiple threads 
      // waiting for items.    
      Monitor.Pulse(listLock); 
     } 
    } 

    public object Consume() 
    { 
     lock (listLock) 
     { 
      // If the queue is empty, wait for an item to be added 
      // Note that this is a while loop, as we may be pulsed 
      // but not wake up before another thread has come in and 
      // consumed the newly added object. In that case, we'll 
      // have to wait for another pulse. 
      while (queue.Count==0) 
      { 
       // This releases listLock, only reacquiring it 
       // after being woken up by a call to Pulse 
       Monitor.Wait(listLock); 
      } 
      return queue.Dequeue(); 
     } 
    } 
} 

Answer 1

코드가 그보다 오래되었습니다. .NET 2.0이 나올 때까지 약간 시간을 썼습니다. 생산자/소비자 큐의 개념하지만 그보다 오래된 방법 :)입니다

네, 코드는 지금까지 내가 알고 있어요으로 안전합니다 -하지만 몇 가지 결함이 있습니다

• 을 그것은 비 제네릭입니다. 현대 버전은 확실하게 일반적인 것입니다.
• 대기열을 중지 할 방법이 없습니다. 대기열을 중지하는 간단한 방법 (모든 소비자 스레드가 종료되도록)은 대기열에 넣을 수있는 "작업 중지"토큰을 갖는 것입니다. 그런 다음 스레드가있는만큼의 토큰을 추가합니다. 또는 중지하려는 플래그가 별도로 있습니다. (대기열의 현재 작업을 모두 마치기 전에 다른 스레드를 중지 할 수 있습니다.)
• 작업이 매우 작 으면 한 번에 하나의 작업을 소비하는 것이 가장 효율적이지 않을 수 있습니다.

솔직히 말해 코드 뒤에있는 아이디어가 코드 자체보다 중요합니다.

Answer 2

다음 코드 조각을 수행 할 수 있습니다. 그것은 일반적이고, 널 (또는 어떤 플래그를 사용하든)을 큐에 넣어서 작업자 스레드가 종료하도록 지시하는 메소드를 가지고 있습니다.

코드는 여기에서 가져온 것입니다 http://www.albahari.com/threading/part4.aspx#_Wait_and_Pulse

using System; 
using System.Collections.Generic; 
using System.Linq; 
using System.Text; 
using System.Threading; 

namespace ConsoleApplication1 
{ 

    public class TaskQueue<T> : IDisposable where T : class 
    { 
     object locker = new object(); 
     Thread[] workers; 
     Queue<T> taskQ = new Queue<T>(); 

     public TaskQueue(int workerCount) 
     { 
      workers = new Thread[workerCount]; 

      // Create and start a separate thread for each worker 
      for (int i = 0; i < workerCount; i++) 
       (workers[i] = new Thread(Consume)).Start(); 
     } 

     public void Dispose() 
     { 
      // Enqueue one null task per worker to make each exit. 
      foreach (Thread worker in workers) EnqueueTask(null); 
      foreach (Thread worker in workers) worker.Join(); 
     } 

     public void EnqueueTask(T task) 
     { 
      lock (locker) 
      { 
       taskQ.Enqueue(task); 
       Monitor.PulseAll(locker); 
      } 
     } 

     void Consume() 
     { 
      while (true) 
      { 
       T task; 
       lock (locker) 
       { 
        while (taskQ.Count == 0) Monitor.Wait(locker); 
        task = taskQ.Dequeue(); 
       } 
       if (task == null) return;   // This signals our exit 
       Console.Write(task); 
       Thread.Sleep(1000);    // Simulate time-consuming task 
      } 
     } 
    } 
} 

Answer 3

경고 : 당신이 코멘트를 읽으면, 내 대답은

이

가능한 교착 상태에있다 : 잘못 이해하게 될 것입니다 당신의 코드.

는 다음과 같은 경우는, 명확성을 위해, 나는 단일 스레드 방식을 사용 상상하지만, 수면 멀티 스레드로 쉽게 변환 할 수 있어야 :이하지 않는 경우 알려 주시기

// We create some actions... 
object locker = new object(); 

Action action1 =() => { 
    lock (locker) 
    { 
     System.Threading.Monitor.Wait(locker); 
     Console.WriteLine("This is action1"); 
    } 
}; 

Action action2 =() => { 
    lock (locker) 
    { 
     System.Threading.Monitor.Wait(locker); 
     Console.WriteLine("This is action2"); 
    } 
}; 

// ... (stuff happens, etc.) 

// Imagine both actions were running 
// and there's 0 items in the queue 

// And now the producer kicks in... 
lock (locker) 
{ 
    // This would add a job to the queue 

    Console.WriteLine("Pulse now!"); 
    System.Threading.Monitor.Pulse(locker); 
} 

// ... (more stuff) 
// and the actions finish now! 

Console.WriteLine("Consume action!"); 
action1(); // Oops... they're locked... 
action2(); 

마십시오 어떤 의미.

이것이 확인되면 질문에 대한 대답은 "아니오, 안전하지 않습니다"입니다.) 도움이되기를 바랍니다.

Answer 4

위의 코드와 위의 article series에서 Monitor.Wait/Pulse가 어떻게 작동하는지 (그리고 일반적으로 스레드에 대해 많이) 배웠습니다. 존 (Jon)이 말했듯이, 그것은 가치가 있으며 실제로 안전하고 적용 가능합니다.

그러나, 현재로.NET 4에는 producer-consumer queue implementation in the framework이 있습니다. 나는 단지 그것을 나 자신을 발견했다. 그러나 지금까지 내가 필요한 모든 것을한다.