c# - 轮询新文件，在处理文件时避免重复

Question

我想从 Windows 服务轮询远程目录(几 MB 的 pdf)中的新文件。

每个文件都必须由密集型 CPU 作业处理(pdf 文件中的图像识别)。该过程完成后，必须将文件移动到其他地方或删除。

我想尽快开始我的工作，同时利用多处理器功能来并行处理工作。

但是，我遇到了一个问题:虽然枚举目录中的文件很容易，但如何避免在我的作业队列中出现重复条目​​？实际上，每次我枚举我的文件时，可能有些文件还在排队或正在处理中。

我的第一个方法是查看 System.Collections.Concurrent.*但似乎没有任何类提供包含方法来在添加之前进行测试。

我也看了HashSet<string> ，但我担心并发访问会出现一些问题。

我目前的骨架是:

    private async void GetNewFiles(CancellationToken cancellationToken)
    {
        if (!cancellationToken.IsCancellationRequested)
        {
            var newfiles = Directory.GetFileSystemEntries(inputDirectory, "*.pdf", SearchOption.AllDirectories);
            logger.Trace($"{newfiles.Length} new files detected in {inputDirectory}");

            foreach (var file in newfiles)
            {
                Task.Factory.StartNew(()=>ProcessFile(file), cancellationToken);
            }

            await Task.Delay(frequency, cancellationToken);
            if (!cancellationToken.IsCancellationRequested)
            {
                GetNewFiles(cancellationToken);
           }
        }
    }

但是，此代码无法避免文件排队两次。

如果我删除了 Task.Delay调用并等待所有文件被处理，它会工作，但它可能导致只有一个正在运行的任务，即使添加了新文件(处理新文件的每次迭代都必须在检查新文件之前完全处理)。

Answer 1

需要对当前代码进行最少修改的最简单方法是使用 ConcurrentDictionary 我认为:

private readonly ConcurrentDictionary<string, byte> _filesInProgress = new ConcurrentDictionary<string, byte>();
private async Task GetNewFiles(CancellationToken cancellationToken) {            
    if (!cancellationToken.IsCancellationRequested) {
        var newfiles = Directory.GetFileSystemEntries(inputDirectory, "*.pdf", SearchOption.AllDirectories);
        foreach (var file in newfiles) {
            // TryAdd returns true if key was not already in dictionary
            if (_filesInProgress.TryAdd(file, 0) && File.Exists(file)) {
                Task.Factory.StartNew(() => {
                    ProcessFile(file);                         
                    _filesInProgress.TryRemove(file, out _);
                }, cancellationToken);
            }
        }
        await Task.Delay(frequency, cancellationToken);
        if (!cancellationToken.IsCancellationRequested) {
            GetNewFiles(cancellationToken);
        }
    }
}

请注意，理想情况下，您希望使用有限数量的线程(等于核心数\虚拟核心数)来处理项目。到目前为止，如果您在目录中找到 100 个文件，您可能会产生 100 个线程，这些线程都是 CPU 密集型的，因此会无缘无故地相互争用资源。