实现并行无限循环的最佳方法是什么？

Question

我已经习惯在 .NET 并行扩展中使用 Parallel.For() ，因为这是一种并行化代码的简单方法，而无需手动启动和维护线程（这可能很繁琐）。我现在正在查看一个无限循环（执行某些操作，直到我发出信号停止），我希望并行化，没有无参数的 Parallel.For() 重载来执行此操作，所以是想知道这里最好的方法是什么。原则上我可以做类似的事情：

Parallel.For(0, int.Max)

但我怀疑这可能不是工作分区逻辑处理的预期/有效模式（？）

另一种选择是这样的：

for(;;)
{
    Parallel.For(0, 128, delegate()
    {
       // Do stuff.
    }
}

但这看起来不优雅，也可能导致工作效率低下分区。

现在我的本能是通过创建和维护自己的线程来手动完成此操作，但我有兴趣获得一些对此的反馈/意见。谢谢。

===更新===

我正在使用接受的答案中文章中的代码的简化版本（我已删除 ParallelOptions 参数）。这是代码...

public class ParallelUtils
{
    public static void While(Func<bool> condition, Action body) 
    { 
        Parallel.ForEach(IterateUntilFalse(condition), ignored => body()); 
    }

    private static IEnumerable<bool> IterateUntilFalse(Func<bool> condition) 
    { 
        while (condition()) yield return true; 
    }
}

示例用法是：

Func<bool> whileCondFn = () => !_requestStopFlag;
ParallelUtils.While(whileCondFn, delegate()
{
    // Do stuff.
});

Answer 1

如果你（真的）想要无限的东西，那么你希望它在尽可能少的核心上。 Parallel.For___ 都不是一个好的选择。

您（可能）需要的是使用 LongRunning 选项创建的单独线程或任务。

然后让它等待信号量，或者作为最后的手段尽可能频繁地调用 Sleep()。

Answer 2

考虑到它是无限微积分请求，但您需要在每个“周期”上有一些有限状态，我想说我会使用外部 for(;;) 循环更改解决方案来执行
Parallel.ForEach(...) 调用某些事件/状态更改。就像 Monitor 信号、event 通知或类似的东西......

Answer 3

这是 ParallelWhile 方法的现代实现，该方法基于 Parallel.For，并且与 Stephen Toub 的实现 从 2009 年开始。

/// <summary>
/// Executes a while loop in which iterations may run in parallel, loop options
/// can be configured, and the state of the loop can be monitored and manipulated.
/// </summary>
public static void ParallelWhile(
    ParallelOptions parallelOptions,
    Func<bool> condition,
    Action<ParallelLoopState> body)
{
    ArgumentNullException.ThrowIfNull(parallelOptions);
    ArgumentNullException.ThrowIfNull(condition);
    ArgumentNullException.ThrowIfNull(body);

    int workersCount = parallelOptions.MaxDegreeOfParallelism switch
    {
        -1 => Int32.MaxValue, // -1 means unlimited parallelism.
        _ => parallelOptions.MaxDegreeOfParallelism
    };

    Parallel.For(0, workersCount, parallelOptions, (_, state) =>
    {
        while (!state.ShouldExitCurrentIteration)
        {
            if (!condition()) { state.Stop(); break; }
            body(state);
        }
    });
}

主要区别在于 Parallel.For 对每个工作线程执行一次迭代，而不是枚举无限序列。然后，每个线程在内部 while 循环中重复调用 body。这样，与包含虚拟元素的序列交互（缓冲/同步等）的开销就被消除了。如果 body 非常轻量级，则此开销可能会很大。

ParallelWhile 支持 Parallel.For 循环的所有选项，并且在发生异常和取消时具有相同的行为。它还保留 的语义MaxDegreeOfParallelism 选项，当它具有默认值-1时，这意味着无限的并行性，实际上意味着“使用所有可用的并行度” ThreadPool 线程”。我总是赞成显式指定 MaxDegreeOfParallelism，并给它一个合理的值。在大多数情况下，此值是 Environment.ProcessorCount。使ThreadPool饱和会产生许多不良副作用，并且几乎没有带来任何好处（如果有的话）。

使用示例：

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

ParallelWhile(options, () => true, state =>
{
    if (TryFindTheAnswerToLifeTheUniverseAndEverything(out _)) state.Stop();
});

Answer 4

我发现了这个技巧
https://dejanstojanovic.net/aspnet/2015/september /parallel-infinite-loop-in-c/

    private static IEnumerable<bool> Infinite()  
    {  
        while (true)  
        {  
            yield return true;  
        }  
    } 
    
private static void test()
{
    Parallel.ForEach(Infinite(), new ParallelOptions(), new Action<bool>((val) =>  
                   {  
                       Console.WriteLine(DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff"));  
                   }));  
          }