BlockingCollection和Parallel.For挂起？

Question

我在玩BlockingCollection，试图更好地理解它们，但我很难理解为什么我的代码在我使用Parallel.For时完成所有项目的处理时挂起。

我只是加了一个数字(制片人？)：

var blockingCollection = new BlockingCollection<long>();

Task.Factory.StartNew(() =>
{
    while (count <= 10000)
    {
        blockingCollection.Add(count);
        count++;
    }
});

然后我试着处理(消费者？)：

Parallel.For(0, 5, x => 
{
    foreach (long value in blockingCollection.GetConsumingEnumerable())
    {
        total[x] += 1;
        Console.WriteLine("Worker {0}: {1}", x, value);
    }
});

但是当它完成所有数字的处理时，它就挂在那里了？我做错了什么？

另外，当我将Parallel.For设置为5时，是否意味着它正在处理5个单独线程上的数据？

Answer 1

顾名思义，当BlockingCollection<T>块不能执行任何操作时，它们就会对它们进行操作，这包括GetConsumingEnumerable()。

这样做的原因是集合无法判断您的生产者是否已经完成，或者只是忙于生产下一个项目。

您需要做的是通知集合，通过调用CompleteAdding()向其添加项。例如：

while (count <= 10000)
{
    blockingCollection.Add(count);
    count++;
}

blockingCollection.CompleteAdding();

Answer 2

这是一个GetConsumingEnumerable方法特性。

如果没有可用项或集合为空，则以这种方式枚举集合将阻止使用者线程。

您可以阅读更多关于它的这里。

而且，使用Parallel.For(0,5)并不能保证数据将在5个单独的线程中处理。这取决于Environment.ProcessorCount。

Answer 3

另外，当我将Parallel.For设置为5时，是否意味着它正在处理5个单独线程上的数据？

不，引用SO(Parallel.For(Foreach)将创建多少线程？默认的MaxDegreeOfParallelism？)中先前的回答：

任务并行库和PLINQ的默认调度程序使用.NET框架ThreadPool来排队和执行工作。在.NET框架4中，ThreadPool使用System.Threading.Tasks.Task类型提供的信息有效地支持并行任务和查询经常表示的细粒度并行性(短暂的工作单元)。

简单地说，TPL创建的是任务，而不是线程。框架决定了应该处理多少线程。