如何优化读取文件列表并将它们存储在数据库中？

Question

我最近在一次采访中被问到一个问题，这真的让我思考。

我正在努力了解和了解有关多线程、并行性和并发性以及性能的更多信息。

场景是您有一个文件路径列表。文件保存在您的 HDD 或 Blob 存储上。 您已阅读文件并将它们存储在数据库中。你会如何以最优化的方式做到这一点？

以下是我能想到的一些方法：

最简单的方法是遍历列表并按顺序执行此任务。

Foreach(var filePath in filePaths)
{
  ProcessFile(filePath);
}

public void ProcessFile(string filePath)
{
  var file = readFile(filePath);
  storeInDb(file);
}

我能想到的第二种方法可能是创建多个线程：

Foreach(var filePath in filePaths)
{
Thread t  = new Thread(ProcessFIle(filePath));
t.Start();
}

(not sure if the above code is correct.)

第三种方式是使用异步等待

List<Tasks> listOfTasks;
Foreach(var filePath in filePaths)
{
  var task = ProcessFile(filePath);
  listOfTasks.Add(task);
}
Task.WhenAll(listOftasks);

public async void ProcessFile(string filePath)
{
  var file = readFile(filePath);
  storeInDb(file);
}

第四种方式是并行。对于：

Parallel.For(0,filePaths.Count , new ParallelOptions { MaxDegreeOfParallelism = 10 }, i =>
    {
        ProcessFile(filePaths[i]);
    });

它们之间有什么区别。哪一个更适合这份工作，还有什么更好的吗？

Answer 1

您还可以使用 Microsoft 的反应式框架（又名 Rx）- NuGet System.Reactive 并添加 using System.Reactive.Linq; - 然后您可以这样做：

IObservable<string> query =
    from filePath in filePaths.ToObservable()
    from file in Observable.Start(() => ReadFile(filePath))
    from db in Observable.Start(() => StoreInDb(file))
    select filePath;

IDisposable subscription =
    query
        .Subscribe(
            filePath => Console.WriteLine($"{filePath} Processed."),
            () => Console.WriteLine("Done."));

Answer 2

我写了一个简单的扩展方法来帮助启动异步任务，限制并发量，并等待它们全部完成；

public static async Task WhenAll(this IEnumerable<Task> tasks, int batchSize)
{
    var started = new List<Task>();

    foreach(var t in tasks)
    {
        started.Add(t);
        if (started.Count >= batchSize)
        {
            var ended = await Task.WhenAny(started);
            started.Remove(ended);
        }
    }
    await Task.WhenAll(started);
}

然后您需要一种将文件内容直接流式传输到数据库的方法。例如；

async Task Process(string filename){
    using var stream = File.OpenRead(filename)

    // TODO connect to the database
    var sqlCommand = ...;
    sqlCommand.CommandText = "update [table] set [column] = @stream";
    sqlCommand.Parameters.Add(new SqlParameter("@stream", SqlDbType.VarBinary)
    {
        Value = stream
    });
    await sqlCommand.ExecuteNonQueryAsync();
}
IEnumerable<string> files = ...;
await files.Select(f => Process(f)).WhenAll(20);

这是最好的方法吗？可能不是。因为这个扩展太容易被滥用了。多次意外启动任务，或一次启动所有任务。