Parallel.For 的 BlockingCollection 挂起？

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

这是一个 GetConsumingEnumerable 方法功能。

Enumerating the collection in this way blocks the consumer thread if no items are available or if the collection is empty.

您可以阅读更多相关信息here

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

Answer 2

Also, when I set my Parallel.For to 5, does it mean it's processing the data on 5 separate thread?

不，引用 SO(How many threads Parallel.For(Foreach) will create? Default MaxDegreeOfParallelism?) 中的先前答案：

The default scheduler for Task Parallel Library and PLINQ uses the .NET Framework ThreadPool to queue and execute work. In the .NET Framework 4, the ThreadPool uses the information that is provided by the System.Threading.Tasks.Task type to efficiently support the fine-grained parallelism (short-lived units of work) that parallel tasks and queries often represent.

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

Answer 3

顾名思义，BlockingCollection<T> 上的操作在无法执行任何操作时会阻塞，这包括 GetConsumingEnumerable().

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

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

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

blockingCollection.CompleteAdding();

Parallel.For 的 BlockingCollection 挂起？

BlockingCollection with Parallel.For hangs?

c#

multithreading

parallel-processing

task-parallel-library

parallel.for