Parallel.ForEach 有序执行

Question

我正在尝试使用新的 C# 4.0 Parallel.ForEach 函数在对象列表上执行并行函数。这是一个非常漫长的维护过程。我想让它按照列表的顺序执行，以便我可以在上一点停止并继续执行。我该怎么做？

这是一个例子。我有一个对象列表：a1 到 a100。这是当前的顺序：

a1, a51, a2, a52, a3, a53...

我想要这个顺序：

a1, a2, a3, a4...

我可以接受某些对象无序运行，但只要我可以在列表中找到一个点，我可以说该点之前的所有对象都已运行。我读了并行编程csharp白皮书，但没有看到任何相关内容。 ParallelOptions 类中没有对此进行设置。

Answer 1

执行如下操作：

int current = 0;
object lockCurrent = new object();

Parallel.For(0, list.Count, 
             new ParallelOptions { MaxDegreeOfParallelism = MaxThreads },
             (ii, loopState) => {
                    // So the way Parallel.For works is that it chunks the task list up with each thread getting a chunk to work on...
                    // e.g. [1-1,000], [1,001- 2,000], [2,001-3,000] etc...
                    // We have prioritized our job queue such that more important tasks come first. So we don't want the task list to be
                    // broken up, we want the task list to be run in roughly the same order we started with. So we ignore tha past in 
                    // loop variable and just increment our own counter.
                    int thisCurrent = 0;
                    lock (lockCurrent) {
                        thisCurrent = current;
                        current++;
                    }
                    dothework(list[thisCurrent]);
                 });

假设您让所有线程在中断之前完成，您可以看到当您退出并行 for 循环时，您将如何知道要执行的最后一个列表项。我不太喜欢 PLINQ 或 LINQ。老实说，我不明白编写 LINQ/PLINQ 如何带来可维护的源代码或可读性...... Parallel.For 是一个更好的解决方案。

Answer 2

如果您使用 Parallel.Break 终止循环，则可以保证低于返回值的所有索引都将被执行。这大约是您能得到的最接近的结果。此处的示例使用 For，但 ForEach 具有类似的重载。

int n = ...
var result = new double[n];

var loopResult = Parallel.For(0, n, (i, loopState) =>
{
   if (/* break condition is true */)
   {
      loopState.Break();
      return;
   }
   result[i] = DoWork(i);
});

if (!loopResult.IsCompleted && 
        loopResult.LowestBreakIteration.HasValue)
{
   Console.WriteLine("Loop encountered a break at {0}", 
                      loopResult.LowestBreakIteration.Value);
}

在 ForEach 循环中，会在内部为每个分区中的每个元素生成迭代索引。执行是无序的，但在中断之后，您知道所有低于 LowestBreakIteration 的迭代都将完成。

摘自“使用 Microsoft .NET 进行并行编程” http://parallelpatterns.codeplex.com/

可在 MSDN 上获取。请参阅 http://msdn.microsoft.com/en-us/library/ff963552 .aspx。 “尽早跳出循环”部分介绍了这种情况。

另请参阅：http://msdn.microsoft.com/en-us/library /dd460721.aspx

Answer 3

对于遇到这个问题的其他人 - 如果您正在循环数组或列表（而不是 IEnumberable ），您可以使用 Parallel.Foreach 的重载，它也提供元素索引来维持原始顺序。

string[] MyArray; // array of stuff to do parallel tasks on 
string[] ProcessedArray = new string[MyArray.Length];
Parallel.ForEach(MyArray, (ArrayItem,loopstate,ArrayElementIndex) =>
{
    string ProcessedArrayItem = TaskToDo(ArrayItem);
    ProcessedArray[ArrayElementIndex] = ProcessedArrayItem;
});

Answer 4

作为替代建议，您可以记录已运行的对象，然后在恢复执行时过滤列表以排除已运行的对象。

如果这需要在应用程序重新启动时保持不变，您可以存储已执行的对象的 ID（我假设这里的对象有一些唯一的标识符）。

Answer 5

对于任何寻找简单解决方案的人，我发布了 2 种扩展方法（一种使用 PLINQ，一种使用 Parallel.ForEach）作为以下问题答案的一部分：

订购 PLINQ ForAll

Answer 6

您似乎正在寻找一个从第一项开始的分区器，一次获取一个项目，直到所有项目都被消耗掉。从 .NET Framework 4.5 开始，标准 .NET 库中已经存在这样的分区器。您只需使用 分区器即可。 Create 方法，使用 EnumerablePartitionerOptions.NoBuffering 选项：

OrderablePartitioner<Item> partitioner = Partitioner.Create(source,
    EnumerablePartitionerOptions.NoBuffering);

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

ParallelLoopResult res = Parallel.ForEach(partitioner, options, (item, state, index) =>
{
    // Process the item
});

每次工作线程完成某个项目的处理时，它都会从 source 集合中获取下一个可用项目。这确保了项目将按大致升序处理。您可以找到 此处有关 Parallel API 使用的各种分区策略（范围分区、块分区等）的更多信息）。

如果您想提前停止并行执行，一种想法是调用 state.Break()。完成 Parallel.ForEach 后，您可以检查属性 ParallelLoopResult.LowestBreakIteration，查看是否调用了 Break，以及最低的 index< /code> 调用了它。

另一个想法是拒绝分区程序对 source 集合元素的无限制访问。作为停止机制，您可以使用 CancellationTokenSource，并作为断路器 TakeWhile LINQ 运算符：

CancellationTokenSource cts = new();
IEnumerable<Item> limited = source.TakeWhile(_ => !cts.IsCancellationRequested);

...然后将 limited 传递给 Partitioner.Create。调用 cts.Cancel() 停止处理。

第三个想法是使用 ParallelOptions.CancellationToken 属性：

ParallelOptions options = new()
{
    MaxDegreeOfParallelism = Environment.ProcessorCount,
    CancellationToken = cts.Token
};

Answer 7

不确定问题是否被改变，因为我的评论似乎是错误的。
这里改进了，基本上提醒并行作业的运行超出了您的控制顺序。
ea 打印 10 个数字可能会得到 1,4,6,7,2,3,9,0。

如果您想停止程序并稍后继续。
类似的问题通常会出现在批处理工作负载中。
并记录所做的事情。
假设您是否必须检查 10.000 个数字是否为素数。
您可以按大小为 100 的批次进行循环，并有素数 log1、log2、log3
log1= 0..99
log2=100..199
请务必设置一些标记来了解批处理作业是否已完成。

这是一个笼统的说法，因为问题也不那么准确。