使用 BlockingCollection.GetConsumableEnumerable 的 Parallel.ForEach 循环

Question

为什么在使用 GetConsumableEnumerable 时，Parallel.ForEach 循环会因 OperationCancelledException 退出？

//outside the function
static BlockingCollection<double> _collection = new BlockingCollection<double>();
    
    
var t = Task.Factory.StartNew(Producer);            
Parallel.ForEach(_collection.GetConsumingEnumerable(),
    item => Console.WriteLine("Processed {0}", item));
Console.WriteLine("FINISHED processing");


public static void Producer()
{
     var data = Enumerable.Range(1, 1000);
     foreach (var i in data)
     {
        _collection.Add(i);
        Console.WriteLine("Added {0}",i);
     }
                    
     Console.WriteLine("Finished adding");
     _collection.CompleteAdding();
}

Answer 1

正如我最近发现的那样，将 Parallel.ForEach 与 BlockingCollection 一起使用有些问题。它可以工作，但需要一些额外的努力。

Stephen Toub 有一篇优秀的博客文章，如果您下载“并行扩展附加”项目（也可在 NuGet 上使用）您会发现一些代码可以为您提供帮助。

Answer 2

使用 Parallel.ForEach与 BlockingCollection作为源，需要进行两项具体调整：

1. 使用 EnumerablePartitionerOptions.NoBuffering 选项。
2. 指定 MaxDegreeOfParallelism 为 -1 以外的值。

正确用法示例：

Partitioner<Item> partitioner = Partitioner.Create(collection.GetConsumingEnumerable(),
    EnumerablePartitionerOptions.NoBuffering);

ParallelOptions options = new()
{
    MaxDegreeOfParallelism = Environment.ProcessorCount
};

Parallel.ForEach(partitioner, options, item =>
{
    //...
});

说明：

1. EnumerablePartitionerOptions .NoBuffering 是必需的，否则 Parallel.ForEach 将不会立即处理每个消耗的项目。相反，它会将项目放入一个小缓冲区中，并等待缓冲区达到任意大小，然后再处理缓冲区中的所有项目。此行为会引入不希望的延迟，甚至可能在某些高级场景中导致死锁。
2. 配置 MaxDegreeOfParallelism需要 才能保留 ThreadPool 使用情况受到控制。否则，Parallel.ForEach 将不断请求越来越多的线程，从而促使 ThreadPool 以每秒一个新线程的速度创建新线程，即使 >BlockingCollection 完全为空，并行循环空闲！有关这种奇怪行为的实验演示，请参阅