如何检测未知并发任务推送和拉取 ConcurrentQueue<T> 的完成情况

Question

几天前，我尝试在我的磁盘上执行快速搜索，但做了一些事情，例如属性、扩展、执行文件内部更改等......

这个想法是让它几乎没有限制/锁定，以避免大文件或目录中包含大量文件等的“延迟”...... 我知道“最佳实践”还很遥远，因为我没有使用诸如“MaxDegreeOfParallelism”之类的东西或带有“while(true)”的拉循环

尽管如此，代码运行得相当快，因为​​我们有支持它的架构。

如果有人想检查发生了什么，我尝试将代码转移到一个虚拟控制台项目。

class Program
{
    static ConcurrentQueue<String> dirToCheck;
    static ConcurrentQueue<String> fileToCheck;
    static int fileCount; //

    static void Main(string[] args)
    {
        Initialize();

        Task.Factory.StartNew(() => ScanDirectories(), TaskCreationOptions.LongRunning);
        Task.Factory.StartNew(() => ScanFiles(), TaskCreationOptions.LongRunning);

        Console.ReadLine();
    }

    static void Initialize()
    {
        //Instantiate caches
        dirToCheck = new ConcurrentQueue<string>();
        fileToCheck = new ConcurrentQueue<string>();

        //Enqueue Directory to Scan here
        //Avoid to Enqueue Nested/Sub directories, else they are going to be dcan at least twice
        dirToCheck.Enqueue(@"C:\");

        //Initialize counters
        fileCount = 0;
    }

    static void ScanDirectories()
    {
        String dirToScan = null;

        while (true)
        {
            if (dirToCheck.TryDequeue(out dirToScan))
            {
                ExtractDirectories(dirToScan);
                ExtractFiles(dirToScan);
            }

            //Just here as a visual tracker to have some kind an idea about what's going on and where's the load
            Console.WriteLine(dirToCheck.Count + "\t\t" + fileToCheck.Count + "\t\t" + fileCount);
        }
    }

    static void ScanFiles()
    {
        while (true)
        {
            String fileToScan = null;
            if (fileToCheck.TryDequeue(out fileToScan))
            {
                CheckFileAsync(fileToScan);
            }
        }
    }

    private static Task ExtractDirectories(string dirToScan)
    {
        Task worker = Task.Factory.StartNew(() =>
        {
            try
            {
                Parallel.ForEach<String>(Directory.EnumerateDirectories(dirToScan), (dirPath) =>
                {
                    dirToCheck.Enqueue(dirPath);
                });

            }
            catch (UnauthorizedAccessException) { }
        }, TaskCreationOptions.AttachedToParent);

        return worker;
    }

    private static Task ExtractFiles(string dirToScan)
    {
        Task worker = Task.Factory.StartNew(() =>
        {
            try
            {
                Parallel.ForEach<String>(Directory.EnumerateFiles(dirToScan), (filePath) =>
                {
                    fileToCheck.Enqueue(filePath);
                });
            }
            catch (UnauthorizedAccessException) { }
        }, TaskCreationOptions.AttachedToParent);

        return worker;
    }

    static Task CheckFileAsync(String filePath)
    {
        Task worker = Task.Factory.StartNew(() =>
        {
            //Add statement to play along with the file here
            Interlocked.Increment(ref fileCount);


            //WARNING !!! If your file fullname is too long this code may not be executed or may just crash
            //I just put a simple check 'cause i found 2 or 3 different error message between the framework & msdn documentation
            //"Full paths must not exceed 260 characters to maintain compatibility with Windows operating systems. For more information about this restriction, see the entry Long Paths in .NET in the BCL Team blog"
            if (filePath.Length > 260)
                return;
            FileInfo fi = new FileInfo(filePath);

            //Add statement here to use FileInfo

        }, TaskCreationOptions.AttachedToParent);

        return worker;
    }
}

问题： 我怎样才能检测到我已经完成了 ScanDirectory？ 完成后，我可以设法将一个空字符串或其他任何内容排入文件队列，以退出它。 我知道如果我使用“AttachedToParent”，我可以在父任务上有一个完成状态，然后例如做一些类似“ContinueWith(()=> { /SomeCode to notice end/} )" 但是父任务仍然在做拉动并且陷入一种无限循环并且每个子语句都开始新的任务。

另一方面，我不能简单地测试每个队列中的“计数”，因为我可能会刷新文件列表和目录列表，但可能还有另一个任务会调用“EnumerateDirectory()”。

我正在尝试找到某种“反应式”解决方案，并避免在循环中出现一些“if()”，因为它是一个简单的带有 AsyncCall 的 while(true){}，因此 80% 的时间都会被检查。

PS：我知道我可以使用 TPL 数据流，我不是因为我被困在 .net 4.0 上，无论如何，在没有数据流的 .net 4.5 中，因为 TPL 几乎没有改进，我仍然很好奇关于它

Answer 1

你可以用BlockingCollection<T>.代替ConcurrentQueue<T>

BlockingCollection<T> 是专门为此类生产者/消费者场景设计的，并提供了一个CompleteAdding 方法，以便生产者可以通知消费者它已完成添加工作。