Необходим шаблон для метода рабочей нити

Question

Мне нужно разработать идеальный метод рабочих потоков. Метод должен выполнить следующие действия:

• 1) извлечь что-то из очереди (скажем очередь строки) и сделать что-то
• 2) прекратить и вернуться, когда класс расположен
• 3) ждать некоторое событие (что очередь не пуста) и не потребляет процессор
• 4) выполняется в отдельном потоке

Основной нить добавит строку очереди и методы потоков сигнала, чтобы продолжить и сделать эту работу.

Я хочу, чтобы вы предоставили мне шаблон с необходимыми объектами синхронизации.

class MyClass, IDisposable 
{ 
    // Thread safe queue from third party 
    private ThreadSafeQueue<string> _workerQueue; 
    private Thread _workerThread; 

public bool Initialize() 
{ 
_workerThread = new Thread(WorkerThread).Start(); 
} 

public AddTask(string object) 
{ 
_workerQueue.Enqueue(object);  
// now we must signal worker thread 
} 

// this is worker thread 
private void WorkerThread() 
{   
    // This is what worker thread must do 
    List<string> objectList = _workerQueue.EnqueAll 
    // Do something  
} 

// Yeap, this is Dispose 
public bool Dispose() 
{ 
} 
} 

Answer 1

То, что вы описываете, лучше всего достигается с моделью производитель-потребитель. Этот шаблон наиболее легко реализуется с блокирующей очередью. Если вы используете .NET 4.0, вы можете воспользоваться классом BlockingCollection. Вот как я вижу, как ваш код работает. В следующем примере я использую значение в качестве дозорного устройства для изящного окончания потребления, но вы также можете воспользоваться параметром CancellationToken по методу Take.

public class MyClass : IDisposable 
{ 
    private BlockingCollection<string> m_Queue = new BlockingCollection<string>(); 

    public class MyClass() 
    { 
    var thread = new Thread(Process); 
    thread.IsBackground = true; 
    thread.Start(); 
    } 

    public void Dispose() 
    { 
    m_Queue.Add(null); 
    } 

    public void AddTask(string item) 
    { 
    if (item == null) 
    { 
     throw new ArgumentNullException(); 
    } 
    m_Queue.Add(item); 
    } 

    private void Process() 
    { 
    while (true) 
    { 
     string item = m_Queue.Take(); 
     if (item == null) 
     { 
     break; // Gracefully end the consumer thread. 
     } 
     else 
     { 
     // Process the item here. 
     } 
    } 
    } 
} 

Answer 2

Я думаю, вы должны рассмотреть возможность использования BackgroundWorker класса, который вполне может соответствовать вашим потребностям.

Answer 3

Звучит как BlockingQueue - это то, что вам нужно.

Answer 4

Вы должны взглянуть на новую .Net 4 System.Collections.Concurrent Namespace. Также this little example должен помочь вам лучше понять, как его использовать.

Answer 5

Попробуйте что-нибудь подобное. экземпляр с типом строки и дать ему делегат обрабатывать вашу строку:

public class SuperQueue<T> : IDisposable where T : class 
{ 
    readonly object _locker = new object(); 
    readonly List<Thread> _workers; 
    readonly Queue<T> _taskQueue = new Queue<T>(); 
    readonly Action<T> _dequeueAction; 

    /// <summary> 
    /// Initializes a new instance of the <see cref="SuperQueue{T}"/> class. 
    /// </summary> 
    /// <param name="workerCount">The worker count.</param> 
    /// <param name="dequeueAction">The dequeue action.</param> 
    public SuperQueue(int workerCount, Action<T> dequeueAction) 
    { 
     _dequeueAction = dequeueAction; 
     _workers = new List<Thread>(workerCount); 

     // Create and start a separate thread for each worker 
     for (int i = 0; i < workerCount; i++) 
     { 
      Thread t = new Thread(Consume) { IsBackground = true, Name = string.Format("SuperQueue worker {0}",i)}; 
      _workers.Add(t); 
      t.Start(); 

     } 

    } 


    /// <summary> 
    /// Enqueues the task. 
    /// </summary> 
    /// <param name="task">The task.</param> 
    public void EnqueueTask(T task) 
    { 
     lock (_locker) 
     { 
      _taskQueue.Enqueue(task); 
      Monitor.PulseAll(_locker); 
     } 
    } 

    /// <summary> 
    /// Consumes this instance. 
    /// </summary> 
    void Consume() 
    { 
     while (true) 
     { 
      T item; 
      lock (_locker) 
      { 
       while (_taskQueue.Count == 0) Monitor.Wait(_locker); 
       item = _taskQueue.Dequeue(); 
      } 
      if (item == null) return; 

      // run actual method 
      _dequeueAction(item); 
     } 
    } 

    /// <summary> 
    /// Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources. 
    /// </summary> 
    public void Dispose() 
    { 
     // Enqueue one null task per worker to make each exit. 
     _workers.ForEach(thread => EnqueueTask(null)); 

     _workers.ForEach(thread => thread.Join()); 

    } 
}