与固定数量的任务/线程并行使用 IEnumerable答案

【问题标题】：Consuming IEnumerable in parallel with a fixed number of Tasks/Threads与固定数量的任务/线程并行使用 IEnumerable
【发布时间】：2014-04-03 12:29:31
【问题描述】：

我有一个源IEnumerable<T>，我想以并行方式处理它，它有固定数量的任务/线程（接近处理器的数量），每个从源中抓取下一个项目并处理它直到所有元素都已迭代。

Parallel.For 不是候选对象，因为元素数量未知。
Parallel.ForEach 不是候选者，因为即使指定 MaxDegreeOfParallelism 也会创建许多任务，因为此参数仅确保最大并发运行的任务数，而不是创建的任务数。
必须通知每个任务源被遍历直到其结束，以便它可以运行一些包装逻辑。
源列表的元素不能保存在内存中，必须不断处理和丢弃。

听起来像是生产者/消费者问题，生产者可以是单线程的，一旦 IEnumerable 完成，将不再添加任何元素。

如何使用 TPL 解决这个问题？我必须实现我自己的可共享线程安全IEnumerable 还是框架提供了什么？

编辑：这是我尝试使用 Parallel.ForEach 并指定 MaxDegreeOfParallelism，这不会阻止 TPL 创建许多任务。

int nbTasks = 0;
Parallel.ForEach(positions, new ParallelOptions { MaxDegreeOfParallelism = Environment.ProcessorCount },
    () => { return new List<IPositionData>(); },
    (position, loop, list) =>
    {
        Thread.Sleep(1);
        list.Add(position);
        return list;
     },
     list => Interlocked.Add(ref nbTasks, 1));
Trace.WriteLine(string.Format("Tasks: {0}", nbTasks));

评论：positions 是我的来源IEnumerable<IPositionData>。我刚刚运行了这个，例如，nbTasks 是 64（而不是我的 4 个内核上的预期 4）。

【问题讨论】：

标签： c# task-parallel-library

【解决方案1】：

您可以通过使用需要ParallelOptions 对象的重载并设置MaxDegreeOfParallelism 属性来限制Parallel.ForEach 中的任务数。

【讨论】：

我已经尝试过了，但没有成功。尽管我将 MaxDegreeOfParallelism 指定为例如 4，但创建的任务比 4 多得多。我猜，这是由于初始化不仅受 CPU 限制。我已经编辑了我的问题以反映这一点。
@Dejan 你能给我们看看你的代码吗？我刚刚尝试了一个简单的示例，我可以验证我的情况 MaxDegreeOfParallelism 是否正常工作。
@Dejan 我看到了你的问题； MaxDegreeOfParallelism 只保证并发任务的数量不超过你的总数，它不限制创建的任务数量。您是否可以修改 localInit lambda 以在后续任务之间重用复杂初始化的结果？然后，您最多必须执行 Environment.ProcessorCount 初始化。如果没有，您可能需要编写自己的TaskScheduler。
或者 - 使用 Parellel.For 但不是在项目数上，而是在您要创建的并发任务数上。然后你只需要同步对你的枚举器的访问。
啊，这很好知道 - 现在我明白了 MaxDegreeOfParallelism，谢谢。但总而言之，这又回到了我最初的问题。我是否必须实现我的一个同步枚举器或自定义调度程序，或者框架是否提供了一些开箱即用的东西来涵盖我的场景。

【解决方案2】：

Parallel.ForEach可以限制任务数量：

   in maxNumberOfTasks = 4;
   Parallel.ForEach(collection, new ParallelOptions { MaxDegreeOfParallelism = maxNumberOfTasks}, 
                 i => {
                        //Your action here
                  });

【讨论】：