.net - AsParallel()和内部缓冲区大小

Question

如何限制AsParallel()预先读取并放入其内部缓冲区的项目数量？

这是一个例子:

int returnedCounter;

IEnumerable<int> Enum()
{
    while (true)
        yield return Interlocked.Increment(ref returnedCounter);
}

[TestMethod]
public void TestMethod1()
{
    foreach (var i in Enum().AsParallel().Select(a => a))
    {
        Thread.Sleep(3000);
        break;
    }
    Console.WriteLine(returnedCounter);
}

我消耗了1项物品， sleep ，请停止枚举。它在我的机器上打印526400。在我的真实项目中，每个项目都分配了数千千字节。 AsParallel()会预先读取很多项，这会导致非常糟糕的内存消耗和CPU浪费。

放入WithMergeOptions(ParallelMergeOptions.NotBuffered)会有所帮助。它打印4544。但是对我来说仍然太多了。

在Enum()中等待将冻结主线程中的循环。

Answer 1

关于Partitioners的另一个问题!

在您的情况下，您将必须查找/编写一次仅需要一项的分区程序。

这是有关Custom Partitioners的文章

更新:

我只记得在哪里看到了SingleItemPartitioner实现:它在此处的ParallelExtensionsExtras项目中:Samples for Parallel Programming with the .NET Framework

我也刚刚阅读了您的测试代码。我可能应该第一次这样做!

这段代码:

Enum().AsParallel().Select(a => a)

意思是:并行获取 Enum()并尽可能快地枚举它，并返回一个新的 IEnumerable<int>。

因此，您的 foreach并不是从 Enum()提取项目，而是从linq语句创建的新 IEnumerable<int>提取项目。

另外，您的 foreach在主线程上运行，因此每个项目上的工作都是单线程的。

如果要并行运行，但仅在需要时生成项，请尝试:

Parallel.ForEach( SingleItemPartitioner.Create( Enum() ), ( i, state ) =>
    {
        Thread.Sleep( 3000 );
        state.Break();
    }