使用 C# 中的多个任务异步处理来自文件的输入记录

Question

我需要使用批处理概念异步处理输入文件中的每条记录。示例 - 假设我的输入文件有 100 条记录，并且我的批量大小设置为 10，我需要分 10 批处理记录（每批 10 条记录），并且所有这些批次都应该异步处理。批量大小 10 不是固定的，可能会有所不同。

我有一个读取函数，它从文件中读取每条记录，如果读取了 10 条记录，我需要调用一个异步方法来处理这些记录 使用任务。但是实际的线程应该继续读取下一组记录并填充下一组记录（读取接下来的 10 条记录），一旦它们被读取，调用相同的异步方法来使用另一个任务处理这些记录，这应该一直持续到所有记录都被阅读。

现在，我可以读取记录并填写批次，然后逐个处理每个批次，但我想异步执行。

我在下面提供了我的代码的 sn-p：

public async Task ProcessRecordAsync(InputFile)

{

   public int recordCount = 0;
   List<Task> TaskList = new List<Task>();

   While (condition to check if records present)

   {
        Object getRecordVal = ReadInputRecord();
       if(++recordCount >= 10)

       {
       var LastTask = new Task(async () => await ProcessRecordAsync());
            LastTask.Start();
            TaskList.Add(LastTask);
       }
    }

    Task.WaitAll(TaskList.ToArray());

 }


ProcessRecordAsync() --> This is the function which processes the input record

我认为我在错误地调用任务时出错了。一旦批次被填充，我想使用任务调用 ProcessRecordAsync 函数，主线程仍应继续读取记录并填充下一组批次。使用此代码，我遇到了异常。

我遇到以下错误：

System.InvalidOperationException：集合已修改；枚举操作可能无法执行。 在 System.ThrowHelper.ThrowInvalidOperationException（ExceptionResource 资源） 在 System.Collections.Generic.List`1.Enumerator.MoveNextRare()

这是处理多项任务的正确方法吗？

Answer 1

您尝试实现的是生产者-消费者模式。实现此模式最熟悉的方法是使用BlockingCollection 类。除了 Add、CompleteAdding 和 GetConsumingEnumerable 方法之外，没有太多要学习的东西。虽然易于学习，但它并不是最强大的工具。它在处理小型工作负载时效率不高，而且由于本质上是阻塞的，因此也不利于可扩展性。也没有对批处理的本机支持。您必须通过管理列表或数组自己完成所有工作。我尝试制作a chunky implementation，但取得了平庸的成功。

最近我花时间学习了TPL Dataflow 库，我可以清楚地说它是适合这项工作的工具。您需要定义两个块，一个用于批处理 (BatchBlock)，另一个用于处理 (ActionBlock)。然后通过将块链接在一起形成一个小管道，将数据提供给第一个块，并由第二个块自动批量处理。最后调用Complete 到第一个块，await 调用第二个块的Completion。该库创建并管理完成这项工作所需的所有Tasks。性能是最佳的。只是 API 不熟悉，不是特别直观，并且当您必须通过为每个构造函数提供选项来配置每个块时有点冗长。

var batchBlock = new BatchBlock<string>(batchSize: 10);
var actionBlock = new ActionBlock<string[]>(batch =>
{
    // Do something with the batch. For example:
    Console.WriteLine("Processing batch");
    foreach (var line in batch)
    {
        Console.WriteLine(line);
    }
});
batchBlock.LinkTo(actionBlock,
    new DataflowLinkOptions() { PropagateCompletion = true });
foreach (var line in File.ReadLines(@".\..\..\_Data.txt"))
{
    await batchBlock.SendAsync(line).ConfigureAwait(false);
}
batchBlock.Complete();
await actionBlock.Completion.ConfigureAwait(false);

Answer 2

我之前没有像其他人建议的那样使用过 MoreLinq 和 TPL DataFlow Library...

如果您想坚持使用 async/await，这样的事情就可以完成（尽管可能会找到一些优化）：

    async Task Main()
    {
        await BatchProcessAsync(GetValues(), ProcessElementAsync);
    }

    public async Task BatchProcessAsync<T>(
        IEnumerable<T> elements,
        Func<T, Task> operationAsync,
        int batchSize = 10)
    {
        using (var en = elements.GetEnumerator())
        {
            var ops = new List<Task>();

            while (en.MoveNext())
            {
                ops.Add(operationAsync(en.Current));
                if (ops.Count == batchSize)
                {
                    await Task.WhenAll(ops);
                    ops.Clear();
                }
            }

            // process any remaining operations
            if (ops.Any()) { await Task.WhenAll(ops); }
        }
    }    

    public async Task ProcessElementAsync(string element)
    {
        Print($"Processing element: {element}...");
        await Task.Delay(300);
        Print($"Completed element: {element}.");

        void Print(string output)
            => Console.WriteLine($"[{DateTime.Now:s}] {output}");
    }

    public IEnumerable<string> GetValues(int maxValues = 100)
        => Enumerable.Range(1, maxValues).Select(i => $"Element #{i}");

编辑发布后，我重新阅读了原始问题并意识到此实现假设您已经阅读了所有记录，因此错过了让主线程继续从输入中读取记录的部分文件。但是，将相同的批处理技术应用于批量读取输入文件中的记录，然后将记录送入进行批处理应该不难。