在 while 循环中实现 Parallel.Foreach

Question

我有需要在 while 循环中运行 Parallel.Foreach 的场景。我需要了解此实现对处理方式的影响。我会有一个类似这样的实现

 ConcurrentQueue<MyTable> queue = new ConcurrentQueue<MyTable>();

在这里，我最初在队列中添加了很多项目，但在执行过程中，还可以在队列中添加更多项目。

while(true)
{
    Parallel.Foreach(queue, (myTable) => {some processing});
    Sleep(sometime);
}

每次一个项目将被取消排队并产生新的线程来处理它，同时将添加新项目，因为我需要保持一个无限的while循环。

现在，我需要了解，由于并发队列是线程安全的，我认为每个项目只会被处理一次，尽管在 foreach 之上，但我不确定是否会有 foreach 本身的多个线程将产生子线程或 foreach 的单个副本将在 while 循环中运行。我不知道 foreach 本身是如何实现的。

Answer 1

我有需要在 while 循环中运行 Parallel.Foreach 的场景。

我认为你不会。您希望处理并行出现的新项目，但我认为这不是最好的方法。

我认为最好的方法是使用来自 TPL Dataflow 的ActionBlock。当没有要处理的项目时，它不会浪费CPU或线程，如果您设置它的MaxDegreeOfParallelism，它将并行处理项目：

ActionBlock<MyTable> actionBlock = new ActionBlock<MyTable>(
    myTable => /* some processing */,
    new ExecutionDataflowBlockOptions
    {
        MaxDegreeOfParallelism = DataflowBlockOptions.Unbounded
    });


...

actionBlock.Post(someTable);

如果您不想或不能（仅限 .Net 4.5）使用 TPL 数据流，另一种选择是使用单个 Parallel.Foreach()（没有 while）与 BlockingCollection 和 GetConsumingPartitioner() （不是 GetConsumingEnumerable()!）。

使用这个，Parallel.Foreach() 线程将在没有要处理的项目时被阻塞，但也不会在处理过程中出现任何延迟（例如由您的Sleep() 引起的延迟）：

BlockingCollection<MyTable> queue = new BlockingCollection<MyTable>();

...

Parallel.ForEach(
    queue.GetConsumingPartitioner(), myTable => /* some processing */);

...

queue.Add(someTable);

---

我认为每个项目只会被处理一次，尽管在 foreach 之上，但我不确定

这就是为什么您应该使用上述选项之一的原因之一，因为它们意味着您不需要了解它们如何工作的详细信息，它们只是工作。