c# - 读取大量文件 "at the same time"

Question

我使用FileSystemWatcher来捕获每个创建、更改、删除和重命名更改文件夹中的任何文件。

对于这些更改，我需要对这些文件的内容执行简单的校验和。简而言之，我打开一个文件流并将其传递给 MD5 类:

private byte[] calculateChecksum(string frl)
{
    using (FileStream stream = File.Open(frl, FileMode.Open, FileAccess.Read, FileShare.ReadWrite))
    {
        return this.md5.ComputeHash(stream);
    }
}

问题在于我需要处理的文件数量。例如，假设我在一个文件夹中创建了 200 个文件，然后我将所有文件复制并粘贴到同一个文件夹中。此操作将导致 200 个事件和 200 个 calculateChecksum() 执行。

如何解决此类问题？

Answer 1

在FileSystemWatcher处理程序中将任务放入将由某些工作人员处理的队列中。工作人员可以以目标速度或/和频率处理校验和计算任务。可能一名工作人员会更好，因为许多读者可以通过多次读取搜索来减慢硬盘驱动器的速度。

尝试阅读有关 BlockingCollection 的内容: https://msdn.microsoft.com/ru-ru/library/dd997371(v=vs.110).aspx

和生产者-消费者数据流模式 https://msdn.microsoft.com/ru-ru/library/hh228601(v=vs.110).aspx

var workerCount = 2;
BlockingCollection<String>[] filesQueues= new BlockingCollection<String>[workerCount];

for(int i = 0; i < workerCount; i++)
{
    filesQueues[i] = new BlockingCollection<String>(500);

    // Worker
    Task.Run(() => 
    {
        while (!filesQueues[i].IsCompleted)
        {
            string url;

            try
            {
                url= filesQueues[i].Take();
            }
            catch (InvalidOperationException) { }

            if (!string.IsNullOrWhiteSpace(url))
            {
                calculateChecksum(url);
            }
        }
    }
}

//FileSystemWatcher 处理程序内部

    var queueIndex = hash(filename) % workersCount
    // Warning!!
    // Blocks if numbers.Count == dataItems.BoundedCapacity
    filesQueues[queueIndex].Add(fileName);
    filesQueues[queueIndex].CompleteAdding();

您还可以创建多个消费者，只需同时调用 Take 或 TryTake - 每个项目只会被一个消费者消费。但请考虑到，在这种情况下，一个文件可以由多个工作人员处理，并且多个硬盘读取器可能会减慢硬盘速度。

UPD 如果有多个工作人员，最好创建多个 BlockingCollections，并将文件推送到索引为的队列中: