如何：Parallel.Foreach 执行许多进程，每个进程运行一个新进程后（但一次一个）？ [关闭]

Question

我相信有人知道这一点，我将非常感谢您的回答。我对委托和异步等知之甚少 - 所以请给我一个如何实现的一般示例。

我有一个工作流程，其中 我可以使用 Parallel.Foreach 同时为许多不同的文件执行一个方法（甜，研磨那个处理器） - 但是在该方法结束后，我需要运行另一种方法（它会生成关于前一个进程的报告）， 并且第二种方法不能并行运行。

我不想在生成报告之前等待 Parallel.ForEach 中的所有文件完成（这不是必需的）。但是，如果我在第一种方法结束时开始报告生成方法，那么我就会遇到问题。有什么排队之类的吗？一定有一些漂亮的方法，对吧？

谢谢

Answer 1

我认为 Jim G 的意思是：

var lockObj = new object();

Parallel.Foreach(files, file => 
{
    // Processing file
    lock(lockObj)
    {
        // Generate report.
    }
});

Answer 2

第二种方法应该链接为continuation task。

在第二种方法中，使用lock 或互斥锁以确保它不会并行运行。

Answer 3

并发和协调运行时 (CCR) 中的 Arbiter.Interleave 协调原语提供了一种实现所需功能的简单方法。基本上你传递了 3 个接收器组，1 个用于并发任务，1 个用于独占任务（不并行执行），1 个用于关闭整个进程。你可以找到一个如何使用它的例子here

Answer 4

另一种选择是使用生产者-消费者模型。您有一个thread safe blocking collection，您将完成的数据放入其中，然后您有一个线程来运行从该集合中提取数据的报告。

//Collection to hold the data the processed files generated
var proccesedDataItems = new new BlockingCollection<ResultData>();

//A thread that processes the files
var processReports = new Task(() =>
{
    //Removes items from the collection, if the collection is empty it blocks
    // or if "CompletedAdded" has been called it will reach the "end" of the 
    // collection
    foreach(var processedData in proccesedDataItems.GetConsumingEnumerable())
    {
        BuildReport(processedData);
    }
});
processReports.Start();    

//Generating the data
Parallel.Foreach(files, file => 
{
   var proccesedData = ProcessFile(file)
   proccesedDataItems.Add(processedData);
});

//Let anyone consuming the collection that you can stop waiting for new items.
proccesedDataItems.CompleteAdding();

Answer 5

这样的事情非常适合 TPL 数据流模型：您创建一个处理文件的并行块，然后创建另一个生成报告的非并行块：

var processFileBlock = new TransformBlock<File, Result>(
    file => ProcessFile(file),
    new ExecutionDataflowBlockOptions
    {
        MaxDegreeOfParallelism = DataflowBlockOptions.Unbounded
    });

var generateReportBlock = new ActionBlock<Result>(
    result => GenerateReport(result));

processFileBlock.LinkTo(generateReportBlock);

foreach (var file in files)
    processFileBlock.Post(file);

如果您还想等到所有处理完成，则需要使用Complete() 和Completetion 添加一些代码。