如何使用任务完成源答案

【问题标题】：How to use TaskCompletionSource如何使用任务完成源
【发布时间】：2018-12-29 09:07:07
【问题描述】：

我遇到了多生产者和单一消费者的情况。我选择了一个公共thread-safe 资源，其中所有生产者Enqueue 项目。但是，当 Dequeue-ing 来自该资源时，我不知道如何有效地使生产者 await 用于新项目。

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");
            }
        }
    }

正如您在Consumerclass 中看到的那样，我想包装我的Storage's methodDequeuein aTaskso that i canawaitit in myConsumer. The only reason i usedTaskCompletionSourceis to be able to communicate between theDequeueandEnque@存储`。

我不知道我是否需要重新初始化 tcs，但我想我需要这样做，因为在每次 Enqueue 操作后我想要一个新的 Task。

我还重新初始化了lock 中的tcs，因为我希望该特定实例设置结果。

我应该如何处理这个？实施好吗？ System.Reactive 会提供更好的选择吗？

【问题讨论】：

生产者/消费者模式有很多选择，试试Blocking Collection、TPL-Dataflow、RX或Channels

标签： asynchronous .net-core task-parallel-library taskcompletionsource

【解决方案1】：

我认为您的实施存在几个问题：

如果你多次调用Enqueue()而不调用DequeueAsync()，你将失去TaskCompletionSource，只有最后一个。然后，在第一次调用后调用DequeueAsync() 将不会产生正确的结果。

要解决此问题，您需要一个 TaskCompletionSource 队列。看看here，再看看here。

如果队列为空，最好在DequeueAsync 中使用SemaphoreSlim.WaitAsync() 以正确等待。

您也没有保护DequeueAsync() 中的队列。

【讨论】：