C#并行foreach同样完成任务答案

【问题标题】：C# parallel foreach equally finishing tasksC#并行foreach同样完成任务
【发布时间】：2013-02-21 12:30:07
【问题描述】：

我正在使用 C# Parallel.ForEach 处理一千多个数据子集。一组需要 5-30 分钟来处理，具体取决于组的大小。在我的电脑里有选项

ParallelOptions po = new ParallelOptions();
po.MaxDegreeOfParallelism = Environment.ProcessorCount

我将获得 8 个并行进程。据我了解，进程在并行任务之间平均分配（例如，第一个任务获得作业编号 1、9、17 等，第二个任务获得 2、10、18 等）；因此，一项任务可以比其他任务更快地完成自己的工作。因为这些数据集比其他数据集花费的时间更少。

问题是四个并行任务在 24 小时内完成了它们的工作，但最后一个在 48 小时内完成。是否有机会组织并行性以使所有并行任务均等完成？这意味着所有并行任务继续工作，直到所有工作都完成？

【问题讨论】：

最长的任务需要多长时间？
最长单循环最长 30 分钟，最长并行任务的总处理时间约为 48 小时。

标签： c# parallel-processing

【解决方案1】：

由于作业不相等，因此您不能在处理器之间分配作业数量并让它们大约在同一时间完成。我认为你需要的是 8 个工作线程来检索下一个工作。您必须使用锁定功能才能获得下一份工作。

如果我错了，请有人纠正我，但在我脑海中......一个工作线程可以被赋予这样的功能：

public void ProcessJob()
{
    for (Job myJob = GetNextJob(); myJob != null; myJob = GetNextJob())
    {
        // process job
    }
}

获得下一份工作的函数如下所示：

private List<Job> jobs;
private int currentJob = 0;

private Job GetNextJob()
{
    lock (jobs)
    {
        Job job = null;
        if (currentJob < jobs.Count)
        {
            job = jobs[currentJob];
            currentJob++;
        }
        return job;
    }
}

【讨论】：

+1 我不确定代码是否正确，但在工作人员空闲时将工作从队列中移除的想法绝对是正确的。

【解决方案2】：

似乎没有现成的解决方案，必须创建它。

我之前的代码是：

var ListOfSets = (from x in Database
           group x by x.SetID into z
           select new { ID = z.Key}).ToList();

ParallelOptions po = new ParallelOptions();
po.MaxDegreeOfParallelism = Environment.ProcessorCount;

Parallel.ForEach(ListOfSets, po, SingleSet=>
{
     AnalyzeSet(SingleSet.ID);
});

为了在所有 CPU 之间平等分担工作，我仍然使用 Parallel 来完成工作，但我使用 For 和 Matt 的想法代替了 ForEach。新代码是：

Parallel.For(0, Environment.ProcessorCount, i=>
{
    while(ListOfSets.Count() > 0)
    {
        double SetID = 0;
        lock (ListOfSets)
        {
            SetID = ListOfSets[0].ID;
            ListOfSets.RemoveAt(0);
        }
     AnalyzeSet(SetID);
    }
});

所以，谢谢你的建议。

【讨论】：

感谢分享您的解决方案。我不怎么做多线程编程，但下次我一定会去看看 Parallel.For。
感谢您的解决方案。注意：在创建 .Count() 和锁定另一个线程的时间之间，由于多线程，可能会占用最后一个列表项。如果您的清单上还有东西，您应该再次检查您的锁，如果没有，则退出。

【解决方案3】：

正如其他人所建议的那样，一种选择是管理您自己的生产者消费者队列。我想指出，使用BlockingCollection 使这非常变得容易。

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

//add data to queue; if it can be done quickly, just do it inline.  
//If it's expensive, start a new task/thread just to add items to the queue.
foreach (JobData job in data)
    queue.Add(job);

queue.CompleteAdding();

for (int i = 0; i < Environment.ProcessorCount; i++)
{
    Task.Factory.StartNew(() =>
    {
        foreach (var job in queue.GetConsumingEnumerable())
        {
            ProcessJob(job);
        }
    }, TaskCreationOptions.LongRunning);
}

【讨论】：