Parallel.Foreach C# 暂停和停止函数？

Question

暂停和停止（在结束之前）parallel.foreach 的最有效方法是什么？

Parallel.ForEach(list, (item) =>
{
    doStuff(item);
});

Answer 1

Damien_The_Unbeliver 有一个很好的方法，但前提是您想让某些外部进程停止循环。如果您想让循环像在正常的 for 或 foreach 循环中使用 break 一样中断，您将需要使用 具有 href="http://msdn.microsoft.com/en-us/library/system.threading.tasks.parallelloopstate.aspx" rel="nofollow noreferrer">ParallelLoopState 作为一个循环体的参数。 ParallelLoopState 有两个与您想要执行的操作相关的函数，Stop() 和 Break()。

函数 Stop() 将在系统方便的时候尽早停止处理元素，这意味着在调用 Stop() 后可以执行更多迭代，并且不能保证处理元素在您停止的元素甚至开始处理之前出现。

函数 Break() 的执行方式与 Stop() 完全相同，但它也会评估位于该项目之前的 IEnumerable 的所有元素。您调用了 Break()。当您不关心元素的处理顺序，但必须处理所有元素直到停止点时，这非常有用。

检查从 foreach 返回的 ParallelLoopResult查看 foreach 是否提前停止，以及是否使用了 Break()，它处理的最小编号的项目是什么。

Parallel.ForEach(list, (item, loopState) =>
    {
        bool endEarly = doStuff(item);
        if(endEarly)
        {
            loopState.Break();
        }
    }
    );
//Equivalent to the following non parallel version, except that if doStuff ends early
//    it may or may not processed some items in the list after the break.
foreach(var item in list)
{
    bool endEarly = doStuff(item);
    if(endEarly)
    {
        break;
    }
}

---

这是一个更实际的例子，

static bool[] list = new int[]{false, false, true, false, true, false};

long LowestElementTrue()
{
    ParallelLoopResult result = Parallel.ForEach(list, (element, loopState) =>
    {
        if(element)
            loopState.Break();
    }
    if(result.LowestBreakIteration.IsNull)
        return -1;
    else
        return result.LowestBreakIteration.Value;
}   

无论它如何分配工作，它总是返回 2 作为答案。

假设处理器调度两个线程来处理此问题，第一个线程处理元素 0-2，第二个线程处理元素 3-5。

Thread 1:                Thread 2
0, False, continue next  3, False, continue next
1, False, continue next  4, True, Break
2, True, Break           5, Don't process Broke

现在，调用的最低索引 Break 是 2，因此 ParallelLoopResult.LowestBreakIteration 每次都会返回 2，无论线程如何分解，因为它始终会处理到数字 2。

这里是一个示例如何使用停止。

static bool[] list = new int[]{false, false, true,  false, true, false};

long FirstElementFoundTrue()
{
    long currentIndex = -1;
    ParallelLoopResult result = Parallel.ForEach(list, (element, loopState, index) =>
    {
        if(element)
        {
             loopState.Stop();

             //index is a 64 bit number, to make it a atomic write
             // on 32 bit machines you must either:
             //   1. Target 64 bit only and not allow 32 bit machines.
             //   2. Cast the number to 32 bit.
             //   3. Use one of the Interlocked methods.
             Interlocked.Exchange (ref currentIndex , index);
        }
    }
    return currentIndex;
}   

根据它如何分配工作，它将返回 2 或 4 作为答案。

假设处理器调度两个线程来处理此问题，第一个线程处理元素 0-2，第二个线程处理元素 3-5。

Thread 1:                 Thread 2
0, False, continue next    3, False, continue next
1, False, continue next    4, True, Stop
2, Don't process, Stopped  5, Don't process, Stopped

在本例中，它将返回 4 作为答案。让我们看看相同的过程，但它是否处理所有其他元素而不是 0-2 和 3-5。

Thread 1:                   Thread 2
0, False, continue next     1, False, continue next
2, True, Stop               3, False, continue next
4, Don't process, Stopped   5, Don't process, Stopped

这次它将返回 2 而不是 4。

Answer 2

为了能够停止 Parallel.ForEach，您可以使用接受 ParallelOptions 参数，并包含一个 CancellationToken 在这些选项中。

有关详细信息，请参阅取消。

至于暂停，一般来说，我不明白你为什么要这样做。您可能正在寻找Barrier（用于协调多个线程之间的工作，比如说它们是否都需要在继续 B 部分之前完成 A 部分），但我认为您不会将其与 Parallel.ForEach 一起使用，因为您不这样做不知道会有多少参与者 是。