我遇到了多生产者和单一消费者的情况。我选择了一个公共thread-safe
资源,其中所有生产者Enqueue
项目。但是,当使用该资源时,我不知道如何有效地制作await
新项目的生产者。Dequeue
POCO
struct Sample
{
public int Id { get; set; }
public double Value { get; set; }
}
生产者
class ProducerGroup
{
StorageQueue sink;
int producerGroupSize;
public ProducerGroup(StorageQueue _sink,int producers)
{
this.sink = _sink;
this.producerGroupSize = producers;
}
public void StartProducers()
{
Task[] producers = new Task[producerGroupSize];
int currentProducer;
for (int i = 0; i < producerGroupSize; i++)
{
currentProducer = i;
producers[i] = Task.Run(async () =>
{
int cycle = 0;
while (true)
{
if (cycle > 5)
{
cycle = 0;
}
Sample localSample = new Sample { Id = currentProducer, Value = cycle++ };
await Task.Delay(1000);
this.sink.Enqueue(localSample);
}
});
}
}
}
贮存
class StorageQueue
{
private TaskCompletionSource<Sample> tcs;
private object @lock = new object();
private Queue<Sample> queue;
public static StorageQueue CreateQueue(int?size=null)
{
return new StorageQueue(size);
}
public StorageQueue(int ?size)
{
if (size.HasValue)
{
this.queue = new Queue<Sample>(size.Value);
}
else
this.queue = new Queue<Sample>();
}
public void Enqueue(Sample value)
{
lock (this.@lock)
{
this.queue.Enqueue(value);
tcs = new TaskCompletionSource<Sample>();
tcs.SetResult(this.queue.Peek());
}
}
public async Task<Sample> DequeueAsync()
{
var result=await this.tcs.Task;
this.queue.Dequeue();
return result;
}
}
消费者
class Consumer
{
private StorageQueue source;
public Consumer(StorageQueue _source)
{
this.source = _source;
}
public async Task ConsumeAsync()
{
while (true)
{
Sample arrivedSample = await this.source.DequeueAsync(); //how should i implement this ?
Console.WriteLine("Just Arrived");
}
}
}
正如您在Consumer
课堂上看到的那样,我想包装我的Storage's method
Dequeue in a
Task so that i can
await it in my
Consumer .
The only reason i used
TaskCompletionSource is to be able to communicate between the
Dequeue and
Enqueue methods in the
Storage`。
我不知道我是否需要重新初始化,tcs
但我想我这样做是因为我Task
在每次Enqueue
操作后都想要一个新的。
我还重新初始化了tcs
内部,lock
因为我希望该特定实例设置结果。
我应该如何处理这个?实施好吗?会System.Reactive
提供更好的选择吗?