更好的 C# 设计以异步遍历文件系统以每天处理大量文件

Question

我正在创建一个 c# 控制台应用程序，它将遍历给定文件夹（和子文件夹）以加密所有文件（二进制或文本）并更新 sqlserver 数据库中的IsEncrypted 标志。客户端上将有数百万个文件需要加密。我们计划在每天的非工作时间（例如从每晚 10 点开始运行 8 小时）按计划任务运行应用程序。

我有两个选择：

选项 1

使用Parallel.ForEach 处理文件。

public void Process(ProcessorOptions options, ProcessorParameter parameter)
{
    int counter = 0;
    CancellationTokenSource cts = new CancellationTokenSource();
    ParallelOptions parallelOptions = new ParallelOptions();
    parallelOptions.CancellationToken = cts.Token;

    try
    {
        parallelOptions.MaxDegreeOfParallelism = Environment.ProcessorCount;
        if (options.NumberOfThreads > 0)
        {
            parallelOptions.MaxDegreeOfParallelism = options.NumberOfThreads;
        }

        if (options.StopTime != 0)
        {
            Timer timer = new Timer(callback => { cts.Cancel(); }, null, options.StopTime * 60000, Timeout.Infinite);
        }

        List<string> storagePaths = parameter.StoragePaths;
        Log("Process Started...");

        foreach (var path in storagePaths)
        {
            Parallel.ForEach(TraverseDirectory(path, f => f.Extension != ".enc"), parallelOptions, file =>
            {
                if (file.Name.IndexOf("SRSCreate.dir") < 0)
                {
                    ProcessFile(parameter, file.FullName, file.Directory.Name, file.Name);
                    counter++;
                }
            });
        }
        Log(string.Format("Process Files Ended... Total File Count = {0}", counter));
    }
    catch (OperationCanceledException ex)
    {
        log.WriteWarningEntry(string.Format("Reached stop time = {0} min, explicit cancellation triggered. Total number of files processed = {1}", options.StopTime, counter.ToString()), ex);
    }
    catch (Exception ex)
    {                
        log.WriteErrorEntry(ex);
    }
    finally
    {
        cts.Dispose();
    }
}

我做了基准测试，发现处理 2000 个文件几乎需要 7-8 分钟。我可以做些什么来提高性能吗？此外，确定下一次运行（第二天）从哪里开始的最佳方法是什么？

选项 2

使用RabbitMQ 的现有设计来推送带有文件路径的消息，以处理文件以实现可扩展性和维护列表。

public void Process(ProcessorOptions options, ProcessorParameter parameter)
{
    try
    {
        using (IConnection connection = parameter.ConnectionFactory.CreateConnection())
        {
            using (IModel channel = connection.CreateModel())
            {
                var queueName = parameter.TopicSubscription.DeriveQueueName();
                var queueDeclareResponse = channel.QueueDeclare(queueName, true, false, false, null);
                EventingBasicConsumer consumer = new EventingBasicConsumer(channel);

                consumer.Received += (o, e) =>
                {
                    string messageContent = Encoding.UTF8.GetString(e.Body);
                    FileData message = JsonConvert.DeserializeObject(messageContent, typeof(FileData)) as FileData;
                    ProcessFile(parameter, message.EntityId, message.Attributes["Id"], message.Attributes["filename"]);
                };

                string consumerTag = channel.BasicConsume(queueName, true, consumer);
            }
        }
    }
    catch (Exception ex)
    {
        log.WriteErrorEntry(ex);
    }
    finally
    {
        Trace.Exit(method);
    }
}

在配置StopTime 之后，我仍然需要弄清楚如何停止阅读消息。性能不是很好，我看到处理 2000 个文件大约需要 25 - 30 分钟。我们认为我们可以在一台机器或多台机器上运行应用程序的多个副本来处理单个队列以进行扩展。您认为，我可以更改此代码以使其更优化吗？

最后一个问题：您认为是否还有其他选项比上述选项更高效和可扩展？

注意：

1) 方法ProcessFile 调用加密逻辑和更新数据库的逻辑。

2）我们遍历文件夹而不是从数据库开始，因为文件系统中可能存在数据库中尚不存在的文件。

Answer 1

这属于性能问题的范畴，所以我将首先链接性能咆哮：https://ericlippert.com/2012/12/17/performance-rant/

这个操作本质上应该是 Diskbound，而不是 CPU bound。进程迭代文件的速度以及读取、加密和写入文件的速度有多快 - 都显然是磁盘绑定的。在磁盘上同时进行更多操作会使速度变慢，而不是变快。当然，除非你有一些极端的设置，比如 SSD 的 Raid 0。

如果有什么可以从多任务处理中受益，那应该是数据库访问。通常那些会通过网络堆栈，特别是如果数据库在另一台计算机上，它很有可能会比磁盘慢。同时，您不想通过查询向数据库发送垃圾邮件。所有查询都有开销，1 200 行查询比 200 1 行查询快。因此，以某种形式的枚举或流式方法获取数据库数据，然后遍历文件。但是哪一个真正最慢取决于每次运行时有多少新的/未加密的文件。

将整个东西移入数据库是可行的。有两种方法可以将 BLOBS 与 DB 一起存储，听起来您正在使用“存储在磁盘上，仅在 DB 中链接”。如果是这样，像 Filestream 这样的属性可能会对您有所帮助：https://www.red-gate.com/simple-talk/sql/learn-sql-server/an-introduction-to-sql-server-filestream/

有点偏离主题，但我的一个 Pet-Peeve 是异常处理，你的示例代码中有一个大罪：

catch (Exception ex)
{
    log.WriteErrorEntry(ex);
}

你捕捉到Exception 但不要让它继续，这意味着你在致命异常之后继续。那只会给你更多——更难理解的——后续例外。所以你永远不应该那样做。有两篇关于异常处理的文章，我确实链接了很多，我认为它们可能对您有所帮助：

• https://blogs.msdn.microsoft.com/ericlippert/2008/09/10/vexing-exceptions/
• https://www.codeproject.com/Articles/9538/Exception-Handling-Best-Practices-in-NET

Answer 2

我不确定生产中会涉及多少物理驱动器。但如果需要，客户可以添加更多。未加密的文件被同一服务器上的加密文件替换，100％的文件需要加密，因为未加密的文件存在安全风险，并且每天，计数都会下降。是的，加密要求文件在内存中才能运行算法。文件的平均大小约为 3 mb。我知道文件大小没有限制，但通常我们会得到巨大的图像文件、word 和 excel doc，然后是一些小的文本文件。

我可以看到问题有很多未知数，这表明单个配置不会在所有情况下都解决问题。所以我的建议是使系统灵活。我将从为所涉及的物理驱动器创建配置开始。每个物理驱动器都应该有一个并发设置。 SSD 驱动器可以在 2-3 个线程同时读取或写入的情况下以最佳方式工作，而硬盘驱动器可能会遇到多个线程。下一个重要的设置是加密线程的数量。理想情况下，当正在运行的线程数等于机器的可用处理器/内核数时，系统应该工作得最好。执行流程应该是这样的：

• IO 线程正在从/向其关联的物理驱动器读取或写入文件。
• 当 IO 线程完成对未加密文件的读取后，会将其排入全局队列以进行处理。
• 当 IO 线程完成写入加密文件时，它还会更新数据库。
• 加密器线程不断汇集全局队列以供文件处理。
• 当加密线程完成对文件的处理后，它会将其排入文件物理驱动器的专用队列中。
• 当 IO 线程空闲时，如果有任何已处理的文件，它会查看其关联物理驱动器的专用队列。如果有，它会将其出列并将其写入磁盘。如果没有，它会继续从磁盘读取另一个文件。

所有这些都可以通过线程或任务以及BlockingCollection 类来实现。不需要Parallel.ForEach 或第三方库。