带有 BlockingCollection.GetConsumableEnumerable 的 Parallel.ForEach 循环

Posted 2023-05-07

【中文标题】带有 BlockingCollection.GetConsumableEnumerable 的 Parallel.ForEach 循环

【英文标题】：Parallel.ForEach loop with BlockingCollection.GetConsumableEnumerable

【发布时间】：2011-09-30 06:50:22

【问题描述】：

为什么Parallel.ForEach 循环以OperationCancelledException 退出，而使用GetConsumableEnumerable？

//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();

【参考方案1】：

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

Stephen Toub 有一个 excellent blog post on it，如果您下载 "Parallel Extension Extras" 项目 (also available on NuGet)，您会发现一些可以帮助您的代码。

【讨论】：

真正让我困惑的是为什么当我调用 _collection.CompleteAdding() 时 Parallel.ForEach 会抛出异常。

@Sam：老实说，我不想说。那里有太多深奥的魔法，我没有信心说出正确的话:)

Parallel Extensions Extras 的当前 URL：code.msdn.microsoft.com/ParExtSamples 并且有人对扩展进行了 NuGet：nuget.org/packages/MSFT.ParallelExtensionsExtras

Can Bilgin 最近关于这种组合的文章 - link

博客链接：ParallelExtensionsExtras Tour – #4 – BlockingCollectionExtensions