需要为工作线程方法模板

Question

我需要设计完美的工作线程方法。该方法必须执行以下操作：

• 1) 从队列中提取一些东西（比如说一个字符串队列）并做一些事情
• 2) 处理类时停止并返回
• 3) 等待某个事件（该队列不为空）并且不消耗cpu
• 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 类。这是我看到您的代码工作的方式。在下面的示例中，我使用null 值作为标记以优雅地结束消费者，但您也可以利用Take 方法上的CancellationToken 参数。

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

试试这样的。用类型字符串实例化并给它一个委托来处理你的字符串：

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

    }
}

Answer 3

您应该看看新的 .Net 4 System.Collections.Concurrent Namespace。另外this little example 应该可以帮助您更好地了解如何使用它。

Answer 4

听起来像BlockingQueue 是你需要的。

Answer 5

我认为您应该考虑使用BackgroundWorker 类，它可能很适合您的需求。