我试图用一组资源实现生产者/消费者模式,因此每个线程都有一个与之相关的资源。例如,我可能有一个任务队列,其中每个任务需要StreamWriter
来编写其结果。每个任务还必须传递给它的参数。生产者 - 消费者使用资源
我从Joseph Albahari的实现开始(参见下面的修改版本)。
我取代的Action
队列与Action<T>
一个队列,其中T
是资源,并通过与螺纹到Action
相关联的资源。但是,这给我带来了如何将参数传递给Action
的问题。显然,Action
必须替换为一个委托,但这会导致在任务排队时如何传递参数(来自ProducerConsumerQueue
类之外)的问题。任何想法如何做到这一点?
class ProducerConsumerQueue<T>
{
readonly object _locker = new object();
Thread[] _workers;
Queue<Action<T>> _itemQ = new Queue<Action<T>>();
public ProducerConsumerQueue(T[] resources)
{
_workers = new Thread[resources.Length];
// Create and start a separate thread for each worker
for (int i = 0; i < resources.Length; i++)
{
Thread thread = new Thread(() => Consume(resources[i]));
thread.SetApartmentState(ApartmentState.STA);
_workers[i] = thread;
_workers[i].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<T> item)
{
lock (_locker)
{
_itemQ.Enqueue(item); // We must pulse because we're
Monitor.Pulse(_locker); // changing a blocking condition.
}
}
void Consume(T parameter)
{
while (true) // Keep consuming until
{ // told otherwise.
Action<T> item;
lock (_locker)
{
while (_itemQ.Count == 0) Monitor.Wait(_locker);
item = _itemQ.Dequeue();
}
if (item == null) return; // This signals our exit.
item(parameter); // Execute item.
}
}
}
问题是我不想为每个任务提供自己的StreamWriter。我希望每个StreamWriter都属于一个线程,它将在执行任务时重用它。 – Johnny 2011-05-14 20:44:06