我对比赛条件有疑问。它们在我编写注释// POSSIBLE RACE的代码示例中概述。这种设计是我自己想出的，但是它存在种族问题，我不确定如何克服这些问题。也许使用信号量是错误的选择。
场景：生产者应该在DB队列中有作业而使用者仍在处理作业时生产作业。如果消费者完成加工工作，则生产者应释放所有消费者，生产者和消费者应退出。
我如何解决下面的问题，这样我就可以拥有一个消费者池和一个生产者，其中生产者向消费者发出信号，如果它们用完了，何时检查更多物品的队列？
我应该使用其他模式吗？我应该使用信号量，互斥量或其他某种锁定机制吗？
谢谢您的帮助！我已经尝试解决这个问题已有一段时间了。
小提琴：https://dotnetfiddle.net/Widget/SeNqQx
public class Producer
{
readonly int processorCount = Environment.ProcessorCount;
readonly List<Consumer> consumers = new List<Consumer>();
ConcurrentQueue<Job> jobs;
readonly object queueLock = new object();
readonly Semaphore producerSemaphore;
readonly Semaphore consumerSemaphore;
public Producer()
{
producerSemaphore = new Semaphore(1, 1);
consumerSemaphore = new Semaphore(processorCount, processorCount);
}
public void StartTask()
{
jobs = GetJobs();
using (var resetEvent = new ManualResetEvent(false))
{
for (var i = 0; i < processorCount; i++)
{
var consumer = new Consumer(jobs, queueLock, producerSemaphore, consumerSemaphore);
consumers.Add(consumer);
QueueConsumer(consumer, processorCount, resetEvent);
}
AddJobsToQueueWhenAvailable(resetEvent);
resetEvent.WaitOne(); // waits for QueueConsumer(..) to finish
}
}
private ConcurrentQueue<Job> GetJobs(){
var q = new ConcurrentQueue<Job>();
for (var i = 0; i < 5; i++) q.Enqueue(new Job()); // this usually comes from DB queue
return q;
}
private void QueueConsumer(Consumer consumer, int numberOfThreadsRunning, ManualResetEvent resetEvent)
{
ThreadPool.QueueUserWorkItem(_ =>
{
try
{
consumer.StartJob();
}
catch (Exception ex)
{
Console.WriteLine("Exception occurred " + ex);
}
finally
{
// Safely decrement the counter
if (Interlocked.Decrement(ref numberOfThreadsRunning) == 0)
{
resetEvent.Set();
}
}
});
}
private void AddJobsToQueueWhenAvailable(ManualResetEvent resetEvent)
{
ThreadPool.QueueUserWorkItem(_ =>
{
while (true) // TODO - replace with cancellation token
{
// lock queue - so that no workers will steal another workers item
lock (queueLock)
{
// check that at least 1 worker is still active
if (consumers.TrueForAll(w => !w.IsRunning))
{
// all jobs complete - release all locks if 0 workers active
consumerSemaphore.Release(processorCount);
return;
}
// poll for new items that have been added to the queue
var newJobs = GetJobs();
// for each item:
foreach (var job in newJobs)
{
// add item to queue
jobs.Enqueue(job);
// If we have any workers halted, let them know there are new items!
if (consumers.Any(w => !w.IsRunning))
{
// POSSIBLE RACE - Consumer may set IsRunning=false, but haven't called wait yet!
// signal worker to continue via semaphore
consumerSemaphore.Release(1);
// wait until worker thread wakes up and takes item before unlocking queue
producerSemaphore.WaitOne();
}
}
} // unlock queue
// sleep for a bit
Thread.Sleep(500); // TODO - replace with cancellation token
}
});
}
}
public class Consumer
{
public bool IsRunning;
ConcurrentQueue<Job> jobs;
private object queueLock;
private Semaphore producerSemaphore;
private Semaphore consumerSemaphore;
public Consumer(ConcurrentQueue<Job> jobs, object queueLock, Semaphore producerSemaphore, Semaphore consumerSemaphore)
{
this.jobs = jobs;
this.queueLock = queueLock;
this.producerSemaphore = producerSemaphore;
this.consumerSemaphore = consumerSemaphore;
}
public void StartJob() {
while(TryGetNextJob(out var job)) {
// do stuff with job
}
}
private bool TryGetNextJob(out Job nextJob)
{
// lock to prevent producer from producing items before we've had a chance to wait
lock (queueLock)
{
if (jobs.TryDequeue(out nextJob))
return true; // we have an item - let's process it
// worker halted
IsRunning = false;
}
// wait for signal from producer
consumerSemaphore.WaitOne();
// once received signal, there should be a new item in the queue - if there is not item, it means all children are finished
var itemDequeued = jobs.TryDequeue(out nextJob);
if (!itemDequeued)
{
return false; // looks like it's time to exit
}
// another item for us to process
IsRunning = true;
// let producer know it's safe to release queueLock
producerSemaphore.Release(); // POSSIBLE RACE - producer may not have locked yet! (WaitOne)
return true;
}
}
public class Job { }
我建议您看看BlockingCollection。但是，许多使用者线程可以调用Take，如果有一个项目将被返回，否则将被阻塞。它还支持在容量上设置限制，以在超出容量时阻止添加线程。
这应该消除对信号量和重置事件的需要，并使代码整体上更简单。 See Blocking Collection and the Producer-Consumer Problem for a more complete description.
感谢您的帮助。我当然会调查BlockingCollection。
所以我实际上离我想要的不远。我只需要阅读更多有关信号量（使用正确的初始计数进行初始化）的代码即可正常工作，以及其他一些细节。搜索EDIT以查看已更改。工作解决方案：
public class Producer
{
readonly int processorCount = Environment.ProcessorCount;
readonly List<Consumer> consumers = new List<Consumer>();
ConcurrentQueue<Job> jobs;
readonly object queueLock = new object();
readonly Semaphore producerSemaphore;
readonly Semaphore consumerSemaphore;
int numberOfThreadsRunning;
public Producer()
{
producerSemaphore = new Semaphore(0, 1); // EDIT - MUST START WITH 0 INITIALLY
consumerSemaphore = new Semaphore(0, processorCount); // EDIT - MUST START WITH 0 INITIALLY
numberOfThreadsRunning = processorCount; // EDIT - take copy so that Interlocked.Decrement references the same int variable in memory
}
public void StartTask()
{
jobs = GetJobs();
using (var resetEvent = new ManualResetEvent(false))
{
for (var i = 0; i < processorCount; i++)
{
var consumer = new Consumer(jobs, queueLock, producerSemaphore, consumerSemaphore);
consumers.Add(consumer);
QueueConsumer(consumer, resetEvent);
}
AddJobsToQueueWhenAvailable(resetEvent);
resetEvent.WaitOne(); // waits for QueueConsumer(..) to finish
}
}
private ConcurrentQueue<Job> GetJobs(){
var q = new ConcurrentQueue<Job>();
for (var i = 0; i < 5; i++) q.Enqueue(new Job()); // this usually comes from DB queue
return q;
}
private void QueueConsumer(Consumer consumer, ManualResetEvent resetEvent)
{
ThreadPool.QueueUserWorkItem(_ =>
{
try
{
consumer.StartJob();
}
catch (Exception ex)
{
lock (queueLock)
{
consumers.Remove(worker);
}
Console.WriteLine("Exception occurred " + ex);
}
finally
{
// Safely decrement the counter
if (Interlocked.Decrement(ref numberOfThreadsRunning) == 0)
{
resetEvent.Set();
}
}
});
}
private void AddJobsToQueueWhenAvailable(ManualResetEvent resetEvent)
{
ThreadPool.QueueUserWorkItem(_ =>
{
while (true) // TODO - replace with cancellation token
{
// lock queue - so that no workers will steal another workers item
lock (queueLock)
{
// check that at least 1 worker is still active
if (consumers.TrueForAll(w => !w.IsRunning))
{
// all jobs complete - release all locks if 0 workers active
consumerSemaphore.Release(processorCount);
return;
}
// poll for new items that have been added to the queue
var newJobs = GetJobs();
// for each item:
foreach (var job in newJobs)
{
// add item to queue
jobs.Enqueue(job);
// If we have any workers halted, let them know there are new items!
if (consumers.Any(w => !w.IsRunning))
{
// POSSIBLE RACE - Consumer may set IsRunning=false, but haven't called wait yet!
// EDIT - Ordering does not matter. If semaphore is Released() before WaitOne() is
// called, then consumer will just continue as soon as it calls WaitOne()
// signal worker to continue via semaphore
consumerSemaphore.Release();
// wait until worker thread wakes up and takes item before unlocking queue
producerSemaphore.WaitOne();
}
}
} // unlock queue
// sleep for a bit
Thread.Sleep(500); // TODO - replace with cancellation token
}
});
}
}
public class Consumer
{
public bool IsRunning;
ConcurrentQueue<Job> jobs;
private object queueLock;
private Semaphore producerSemaphore;
private Semaphore consumerSemaphore;
public Consumer(ConcurrentQueue<Job> jobs, object queueLock, Semaphore producerSemaphore, Semaphore consumerSemaphore)
{
this.jobs = jobs;
this.queueLock = queueLock;
this.producerSemaphore = producerSemaphore;
this.consumerSemaphore = consumerSemaphore;
CurrentlyProcessing = true; // EDIT - must default to true so producer doesn't exit prematurely
}
public void StartJob() {
while(TryGetNextJob(out var job)) {
// do stuff with job
}
}
private bool TryGetNextJob(out Job nextJob)
{
// lock to prevent producer from producing items before we've had a chance to wait
lock (queueLock)
{
if (jobs.TryDequeue(out nextJob))
return true; // we have an item - let's process it
// worker halted
IsRunning = false;
}
// wait for signal from producer
consumerSemaphore.WaitOne();
// once received signal, there should be a new item in the queue - if there is not item, it means all children are finished
var itemDequeued = jobs.TryDequeue(out nextJob);
if (!itemDequeued)
{
return false; // looks like it's time to exit
}
// another item for us to process
IsRunning = true;
// let producer know it's safe to release queueLock
producerSemaphore.Release(); // POSSIBLE RACE - producer may not have locked yet! (WaitOne)
// EDIT - Order does not matter. If we call Release() before producer calls WaitOne(), then
// Producer will just continue as soon as it calls WaitOne().
return true;
}
}
public class Job { }