C# 中的生产者消费者，具有多个（并行）消费者且没有 TPL 数据流

Question

我正在尝试实现具有多个或并行消费者的生产者/消费者模式。

我做了一个实现，但我想知道它有多好。有人可以做得更好吗？你们中的任何人都可以发现任何错误吗？

很遗憾，我不能使用 TPL 数据流，因为我们的项目已经结束，在我们的包中放入一个额外的库需要大量的文书工作，而我们没有时间。

我要做的是加快以下部分的速度：

anIntermediaryList = StepOne(anInputList); // I will put StepOne as Producer :-) Step one is remote call.

aResultList = StepTwo(anIntermediaryList); // I will put StepTwo as Consumer, however he also produces result. Step two is also a remote call.
// StepOne is way faster than StepTwo.

为此，我想出了将输入列表（anInputList）分块的想法

StepOne 将位于生产者内部，并将中间块放入队列中。 会有多个 Producer，他们会取中间结果，用 StepTwo 处理。

这是稍后实现的简化版本：

Task.Run(() => {
     aChunkinputList = Split(anInputList)
     foreach(aChunk in aChunkinputList)
     {
          anIntermediaryResult = StepOne(aChunk)
          intermediaryQueue.Add(anIntermediaryResult)
     }
})

while(intermediaryQueue.HasItems)
{
     anItermediaryResult = intermediaryQueue.Dequeue()
     Task.Run(() => {
         aResultList = StepTwo(anItermediaryResult);
         resultQueue.Add(aResultList)
     }
}

我还认为并行运行的消费者的最佳数字是：“Environment.ProcessorCount / 2”。我想知道这是否也是一个好主意。

现在这是我的模拟实现，问题是有人可以做得更好或发现任何错误吗？

class Example
{
    protected static readonly int ParameterCount_ = 1000;
    protected static readonly int ChunkSize_ = 100;
    // This might be a good number for the parallel consumers.
    protected static readonly int ConsumerCount_ = Environment.ProcessorCount / 2;
    protected Semaphore mySemaphore_ = new Semaphore(Example.ConsumerCount_, Example.ConsumerCount_);

    protected ConcurrentQueue<List<int>> myIntermediaryQueue_ = new ConcurrentQueue<List<int>>();
    protected ConcurrentQueue<List<int>> myResultQueue_ = new ConcurrentQueue<List<int>>();

    public void Main()
    {
        List<int> aListToProcess = new List<int>(Example.ParameterCount_ + 1);
        aListToProcess.AddRange(Enumerable.Range(0, Example.ParameterCount_));

        Task aProducerTask = Task.Run(() => Producer(aListToProcess));

        List<Task> aTaskList = new List<Task>();            
        while(!aProducerTask.IsCompleted || myIntermediaryQueue_.Count > 0)
        {
            List<int> aChunkToProcess;
            if (myIntermediaryQueue_.TryDequeue(out aChunkToProcess))
            {
                mySemaphore_.WaitOne();
                aTaskList.Add(Task.Run(() => Consumer(aChunkToProcess)));
            }
        }

        Task.WaitAll(aTaskList.ToArray());

        List<int> aResultList = new List<int>();
        foreach(List<int> aChunk in myResultQueue_)
        {
            aResultList.AddRange(aChunk);
        }
        aResultList.Sort();
        if (aListToProcess.SequenceEqual(aResultList))
        {
            Console.WriteLine("All good!");
        }
        else
        {
            Console.WriteLine("Bad, very bad!");
        }
    }

    protected void Producer(List<int> elements_in)
    {
        List<List<int>> aChunkList = Example.SplitList(elements_in, Example.ChunkSize_);

        foreach(List<int> aChunk in aChunkList)
        {
            Console.WriteLine("Thread Id: {0} Producing from: ({1}-{2})", 
                Thread.CurrentThread.ManagedThreadId,
                aChunk.First(),
                aChunk.Last());

            myIntermediaryQueue_.Enqueue(ProduceItemsRemoteCall(aChunk));
        }
    }

    protected void Consumer(List<int> elements_in)
    {
        Console.WriteLine("Thread Id: {0} Consuming from: ({1}-{2})",
                Thread.CurrentThread.ManagedThreadId,
                Convert.ToInt32(Math.Sqrt(elements_in.First())),
                Convert.ToInt32(Math.Sqrt(elements_in.Last())));

        myResultQueue_.Enqueue(ConsumeItemsRemoteCall(elements_in));
        mySemaphore_.Release();
    }

    // Dummy Remote Call
    protected List<int> ProduceItemsRemoteCall(List<int> elements_in)
    {
        return elements_in.Select(x => x * x).ToList();
    }

    // Dummy Remote Call
    protected List<int> ConsumeItemsRemoteCall(List<int> elements_in)
    {
        return elements_in.Select(x => Convert.ToInt32(Math.Sqrt(x))).ToList();
    }

    public static List<List<int>> SplitList(List<int> masterList_in, int chunkSize_in)
    {
        List<List<int>> aReturnList = new List<List<int>>();
        for (int i = 0; i < masterList_in.Count; i += chunkSize_in)
        {
            aReturnList.Add(masterList_in.GetRange(i, Math.Min(chunkSize_in, masterList_in.Count - i)));
        }
        return aReturnList;
    }
}

主要功能：

class Program
{
    static void Main(string[] args)
    {
        Example anExample = new Example();
        anExample.Main();
    }
}

再见 拉兹洛

Answer 1

基于我发布的第二个和第三个版本的 cmets： https://codereview.stackexchange.com/questions/71182/producer-consumer-in-c-with-multiple-parallel-consumers-and-no-tpl-dataflow/71233#71233