使用任务处理大型输入文件

Question

2016 年 11 月 14 日更新，因为我看到一些答案（非常感谢）要求提供更多详细信息...

语言：VB.Net / Framework: 4.5

尊敬的 TPL 专家，我是任务处理库 (TPL) 的新手，希望在以下方面提供任何帮助：

我需要处理一个大文件（数十万条记录）。每条记录是一个制表符分隔的记录，最多包含 4 个字段。

对于读取的每条记录，我需要提取字段并调用 asnyc 任务来处理这些字段。

显然，我需要通过一次最多允许处理 10 记录来限制进程，即一次最多启动 10 任务。

例如，如果我读取了第一条10 记录，那么10 任务将被启动。如果任何一项任务完成，我会读取下一条记录并开始另一项任务，依此类推，直到文件结束。基本上，我想优化我可以同时读取和处理的记录数。

到目前为止，这是我想出的：

   rec_list = New StreamReaderEnumerable( file_spec )

   For each rec In rec_list 

      task_list.Add( Task.Run( Async Function()
                                   Return Await task_func( rec )
                               End Function
                              )
                   )

      If ( task_list.Count >= 10 ) Then

         Task.WhenAll( task_list )

         task_list.Clear()

      End If

   Next

“StreamReaderEnumerable”类用于一次从文件中返回一条记录（作为 Enumerable 源），然后我将任务添加到列表中，以便一次处理 10 个任务。

没有必要维护任何类型的顺序，因为每个任务都可能随时完成——我只是想提高一次处理多于一个记录的效率。

问题是，我目前正在等待所有 10 项任务完成，然后才能继续。最好能进一步优化。

我想我正在寻找的是一个 ForEachAsync 类型的枚举器——但还没有找到任何明确的例子......

我们将不胜感激。

Answer 1

1. 你应该看看BlockingCollection class。您可以通过10 限制集合的大小，创建一个消费者任务，它将向集合添加一行，以及一些生产者任务，这将从集合中取出一行直到文件末尾。像这样的：

   var numbers = new BlockingCollection<string>(10);
   // this task should be refactored to accept a method
   Task.Run(() =>
   {
       while (!lines.IsCompleted)
       {
           try
           {
               var line = lines.Take();
               // do stuff here
           }
           catch (InvalidOperationException)
           {
               Console.WriteLine("Adding was completed!");
               break;
           }
       }
       Console.WriteLine("No more lines to take.");
   });
   
   // A simple blocking producer with no cancellation.
   Task.Run(() =>
   {
       while (!END_OF_FILE_CHECK)
       {
           // this method will block until you have less than 10 lines in a collection
           lines.Add(line);
       }
       lines.CompleteAdding();
   });
   

   这两个任务都应该使用LongRunning flag 创建，因为这将为所有任务分配一个线程：

   指定一个任务将是一个长时间运行的粗粒度操作，比细粒度系统涉及更少、更大的组件。它向TaskScheduler 提供了一个提示，即超额订阅可能是有保证的。超额订阅允许您创建比可用硬件线程数更多的线程。它还向任务调度程序提供了一个提示，即该任务可能需要一个额外的线程，这样它就不会阻塞其他线程或本地线程池队列上的工作项的前进。

2. 正如@ScottChamberlain 所建议的，其他方法是使用具有有限并行度的TPL Dataflow 的相同生产者/消费者模式，如下所示：

   var workerBlock = new ActionBlock<string>(
       line =>
       {
           // do stuff here   
       },
       // No more than 10 lines simultaneosly
       new ExecutionDataflowBlockOptions
       {
           MaxDegreeOfParallelism = 10
       });
   
   while (!END_OF_FILE_CHECK)
   {
       workerBlock.Post(line);
   }
   workerBlock.Complete();
   
   // Wait for all messages to propagate through the network.
   workerBlock.Completion.Wait(); 
   

   您也可以引入一些生产者块like here in tutorial，并添加一些链接到worker 的其他块来处理它的结果。