C# Parallel - 将项目添加到被迭代的集合中，还是等效的？答案

【问题标题】：C# Parallel - Adding items to the collection being iterated over, or equivalent?C# Parallel - 将项目添加到被迭代的集合中，还是等效的？
【发布时间】：2015-11-11 20:37:34
【问题描述】：

现在，我有一个 C# 程序，它会重复执行以下步骤：

从数据库中获取当前任务列表
使用 Parallel.ForEach()，为每个任务做工作

但是，其中一些任务运行时间很长。这会延迟其他待处理任务的处理，因为我们只在程序开始时寻找新任务。

现在，我知道修改被迭代的集合是不可能的（对吗？），但是 C# Parallel 框架中是否有一些等效的功能可以让我在处理列表中的项目的同时向列表中添加工作列表？

【问题讨论】：

您是否使用计时器从数据库中抓取任务？如果是这样，请在任务表中添加状态列并将任务标记为“处理中”。任务完成后，将其标记为“完成”。因此，在您的计时器内，您应该只抓取具有“空”状态的任务。关于任务列表，它应该在您的“ProcessTasks”方法中声明。这意味着每次计时器运行时，您都会得到一个新列表。
请原谅，但我不太明白这如何解决问题。我们确实在计时器上抓取任务并具有处理和完成列。问题是，一旦我们获取了任务列表并开始使用 Parallel.ForEach 处理它们，我们就会被困在 ForEach 中，直到所有处理完成。我想我们可以将 Parallel 处理放入 Task() 中，以便主线程可以继续其计时器。
并行应该为每个任务生成一个线程。这意味着如果您有 5 个任务，它们应该被独立处理，并且一旦这 5 个线程启动，您就应该退出循环，对吗？你看过线程安全collections吗？它可能会帮助你。
您确定 ForEach 在启动每个线程后返回吗？我可以发誓它会阻塞，直到所有线程都完成。
@KosalaW Parallel.ForEach 在枚举完每个元素或被中断之前不会完成。它很像传统的foreach 循环，但可以并行执行。 msdn.microsoft.com/en-us/library/dd992001%28v=vs.110%29.aspx

标签： c# parallel-processing parallel.foreach

【解决方案1】：

这是您可以尝试的方法示例。我认为您想摆脱Parallel.ForEaching 并使用异步编程来做一些事情，因为您需要在结果完成时检索结果，而不是在可以想象包含长时间运行的任务和很快完成的任务的离散块中。

此方法使用简单的顺序循环从异步任务列表中检索结果。在这种情况下，您应该可以安全地使用简单的非线程安全可变列表，因为列表的所有突变都在同一个线程中按顺序发生。

请注意，这种方法在循环中使用Task.WhenAny，这对于大型任务列表效率不高，在这种情况下您应该考虑另一种方法。（见此博客：http://blogs.msdn.com/b/pfxteam/archive/2012/08/02/processing-tasks-as-they-complete.aspx）

本示例基于：https://msdn.microsoft.com/en-GB/library/jj155756.aspx

private async Task<ProcessResult> processTask(ProcessTask task) 
{
    // do something intensive with data
}

private IEnumerable<ProcessTask> GetOutstandingTasks() 
{
    // retreive some tasks from db
}

private void ProcessAllData()
{
    List<Task<ProcessResult>> taskQueue = 
        GetOutstandingTasks()
        .Select(tsk => processTask(tsk))
        .ToList(); // grab initial task queue

    while(taskQueue.Any()) // iterate while tasks need completing
    {
        Task<ProcessResult> firstFinishedTask = await Task.WhenAny(taskQueue); // get first to finish
        taskQueue.Remove(firstFinishedTask); // remove the one that finished
        ProcessResult result = await firstFinishedTask; // get the result
        // do something with task result
        taskQueue.AddRange(GetOutstandingTasks().Select(tsk => processData(tsk))) // add more tasks that need performing
    }
}

【讨论】：

非常感谢您的回答，但我认为它没有解决我遇到的核心问题。我不想处理较早完成的结果（实际上，没有后处理要做）。相反，我希望并行处理任务列表，并且在该处理运行时，将更多任务添加到列表中（也可以并行处理）。 await Task.WhenAny 调用将阻塞，直到第一个任务完成。如果该任务是长时间运行的，那么在我们调用 GetOutstandingTasks() 之前将需要很长时间，这正是我现在遇到的问题。
@Eric Task.WhenAny 不会阻塞直到第一个任务完成，它会阻塞直到列表中的 any 任务完成。这意味着您可以在等待长时间运行的任务结果的同时完成和处理大量快速运行的任务。长时间运行的任务在完成之前不会出现。
@Eric 我更新了我的答案，希望现在措辞更清楚。
那么，只有在任务完成时才查询新任务？当问题明确指出某些任务可能需要很长时间时，这对我来说听起来不是一个好方法。例如，如果当前只有一个这样的任务正在处理，您的代码将等待很长时间才能获得新任务。
@svick 这正是我的问题。一些长期运行的任务阻碍了我们处理一些可能短期运行的任务。

【解决方案2】：

一般来说，您是对的，不允许在迭代时修改集合。但是您还可以使用其他方法：

使用ActionBlock<T> from TPL Dataflow。代码可能类似于：

var actionBlock = new ActionBlock<MyTask>(
    task => DoWorkForTask(task),
    new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = DataflowBlockOptions.Unbounded });

while (true)
{
    var tasks = GrabCurrentListOfTasks();
    foreach (var task in tasks)
    {
        actionBlock.Post(task);

        await Task.Delay(someShortDelay);
        // or use Thread.Sleep() if you don't want to use async
    }
}

使用BlockingCollection<T>，可以在使用其中的项目时对其进行修改，同时使用GetConsumingParititioner() from ParallelExtensionsExtras 使其与Parallel.ForEach() 一起使用：

var collection = new BlockingCollection<MyTask>();

Task.Run(async () =>
{
    while (true)
    {
        var tasks = GrabCurrentListOfTasks();
        foreach (var task in tasks)
        {
            collection.Add(task);

            await Task.Delay(someShortDelay);
        }
    }
});

Parallel.ForEach(collection.GetConsumingPartitioner(), task => DoWorkForTask(task));

【讨论】：