동적 대기열에 작업을 대기열에 넣는 방법

Question

다른 작업을위한 대기열을 만들어야합니다.

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

public class PCQueue 
{ 
    readonly object _locker = new object(); 
    Thread[] _workers; 
    Queue<Action> _itemQ = new Queue<Action>(); 

    public PCQueue (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 Shutdown (bool waitForWorkers) 
    { 
    // Enqueue one null item per worker to make each exit. 
    foreach (Thread worker in _workers) 
     EnqueueItem (null); 

    // Wait for workers to finish 
    if (waitForWorkers) 
     foreach (Thread worker in _workers) 
     worker.Join(); 
    } 

    public void EnqueueItem (Action item) 
    { 
    lock (_locker) 
    { 
     _itemQ.Enqueue (item);   // We must pulse because we're 
     Monitor.Pulse (_locker);   // changing a blocking condition. 
    } 
    } 

    void Consume() 
    { 
    while (true)      // Keep consuming until 
    {         // told otherwise. 
     Action item; 
     lock (_locker) 
     { 
     while (_itemQ.Count == 0) Monitor.Wait (_locker); 
     item = _itemQ.Dequeue(); 
     } 
     if (item == null) return;   // This signals our exit. 
     item();       // Execute item. 
    } 
    } 
} 

주요 방법으로 : 현재이 http://www.albahari.com/threading/part4.aspx#_Wait_and_Pulse 의해 제공된 예의 정의 버전으로 수행

static void Main() 
{ 
    PCQueue q = new PCQueue (2); 

    Console.WriteLine ("Enqueuing 10 items..."); 

    for (int i = 0; i < 10; i++) 
    { 
    int itemNumber = i;  // To avoid the captured variable trap 
    q.EnqueueItem (() => 
    { 
     Thread.Sleep (1000);   // Simulate time-consuming work 
     Console.Write (" Task" + itemNumber); 
    }); 
    } 

    q.Shutdown (true); 
    Console.WriteLine(); 
    Console.WriteLine ("Workers complete!"); 
} 

그러나 동안이 수정 된 버전을 통해 발견 유래 탐색 :

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

namespace Project 
{ 
    /// <summary> 
    /// Description of Multithread. 
    /// </summary> 
    public class Multithread<T> : IDisposable where T : class 
    { 
     object locker = new object(); 
     Thread[] workers; 
     Queue<T> taskQ = new Queue<T>(); 

     public void 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 
       System.Diagnostics.Debug.WriteLine(task); 
       Thread.Sleep(1000);    // Simulate time-consuming task 
      } 
     } 
    } 
} 

을

더 나은 사용성을 제공하는 것으로 보입니다. 그러나이 대기열에 작업을 올바르게 추가하는 방법을 찾을 수 없습니다.

testthread.EnqueueTask(testclass.functioname()); 

그러나 작동하지 않는 것 :

classname testclass = new classname(); 
Multithread<classname> testthread = new Multithread<classname>(); 

나는이 라인을 따라 뭔가있을 거라 생각 했어요. 이 문제에 붙어있어이 문제에 대한 도움을 찾을 수 없습니다.

Answer 1

실질적으로 항목을 소비하는 방법을 결정하지 않아서 일반적인 방식으로 생산자/소비자 패턴을 구현하는 방법에 대한 데모가 나타나기 때문에 Multithread이 더 나은 사용법을 제공하는 방법을 알지 못합니다. 반면에 PCQueue은 실제로 항목을 소비하도록 허용하는 작업과 함께 작동합니다.

당신이 제네릭 형식 매개 변수 T를 제거하고 Action에 의해 T의 모든 항목을 대체 할 수있는 몇 가지 작업을 할 수 있도록 Multithread을 수정합니다. Consume 방법에서는 다음 Action를 제공하여 PCQueue를 사용할 때 완료 당신이 정확히 어떻게해야 작업을 대기열하기

task(); 

하여 코드

System.Diagnostics.Debug.WriteLine(task); 
Thread.Sleep(1000);    // Simulate time-consuming task 

를 교체해야합니다. 이를 위해 람다 식을 사용할 수 있습니다.

Answer 2

BlockingCollection을 사용하면이 작업을 훨씬 단순화 할 수 있습니다. 이 데이터 구조는 이미 생산자 - 소비자 논리를 캡슐화 한 대기열로 구현됩니다.

public class PCQueue 
{ 
    private Thread[] workers; 
    private BlockingCollection<Action> queue = new BlockingCollection<Action>(); 
    private CancellationTokenSource cts = new CancellationTokenSource(); 

    public PCQueue(int workerCount) 
    { 
    workers = new Thread[workerCount]; 
    for (int i = 0; i < workerCount; i++) 
    { 
     workers[i] = new Thread(Run); 
     workers[i].Start(); 
    } 
    } 

    public void Shutdown(bool waitForWorkers) 
    { 
    cts.Cancel(); 
    if (waitForWorkers) 
    { 
     foreach (Thread thread in workers) 
     { 
     thread.Join(); 
     } 
    } 
    } 

    public void EnqueueItem(Action action) 
    { 
    queue.Add(action); 
    } 

    private void Consumer() 
    { 
    while (true) 
    { 
     Action action = queue.Take(cts.Token); 
     try 
     { 
     if (action != null) action(); 
     } 
     catch (Exception caught) 
     { 
     // Notify somebody that something bad happened. 
     } 
    } 
    } 
}