C#中的并行编程

Question

我有兴趣学习 C#.NET 中的并行编程（不像所有要知道的东西，而是基础知识和一些好的实践），因此我决定重新编程我的一个旧程序，它是称为 ImageSyncer。 ImageSyncer 是一个非常简单的程序，它所做的只是扫描一个文件夹并找到所有以 .jpg 结尾的文件，然后它根据文件的拍摄日期计算文件的新位置（解析 xif 数据或其他它被称为）。生成位置后，程序会检查该位置是否存在任何现有文件，如果存在，它会查看要复制的文件和“以它的方式”文件的最后写入时间。如果它们相等，则跳过文件。如果不是，则创建并匹配两个文件的 md5 校验和。如果不匹配，则为要复制的文件指定要复制到的新位置（例如，如果要复制到“C:\test.jpg”，则将其复制到“C:\test(1).jpg”。 jpg”代替）。此操作的结果被填充到一个结构类型的队列中，该队列包含两个字符串，即原始文件和将其复制到的位置。然后迭代该队列，直到它为空并复制文件。

也就是说有4个操作：

1. Scan directory for jpegs  
2. Parse files for xif and generate copy-location  
3. Check for file existence and if needed generate new path  
4. Copy files

所以我想重写这个程序以使其并行并能够同时执行多个操作，我想知道实现这一目标的最佳方法是什么。我想出了我能想到的两种不同的模型，但它们中的任何一种都可能根本没有任何好处。第一个是将旧程序的 4 个步骤并行化，以便在执行步骤 1 时在多个线程上完成，当整个步骤 1 完成时，开始步骤 2。另一个（我觉得更有趣，因为我不知道该怎么做）是创建一种工作者和消费者模型，所以当一个线程完成第 1 步时，另一个线程接管并执行第 2 步对象（或类似的东西）。但正如所说，我不知道这些是否有任何好的解决方案。另外，我对并行编程一无所知。我知道如何创建一个线程，以及如何让它执行一个以对象为唯一参数的函数，并且我也曾经使用过 BackgroundWorker 类，但我对它们中的任何一个都不熟悉.

我们将不胜感激。

Answer 1

有几个选项：

• 并行 LINQ：Running Queries On Multi-Core Processors

• 任务并行库 (TPL)：Optimize Managed Code For Multi-Core Machines

• 如果您对基本线程原语和概念感兴趣：Threading in C#

[但正如@John Knoeller 指出的那样，您给出的示例可能是顺序 I/O 绑定的]

Answer 2

这是我用于 C# 线程的参考：http://www.albahari.com/threading/

作为单个 PDF：http://www.albahari.com/threading/threading.pdf

对于您的第二种方法：

我曾开发过一些生产者/消费者多线程应用程序，其中每个任务都是一些永远循环的代码。外部“初始化程序”为每个任务启动一个单独的线程，并为每个任务初始化一个 EventWaitHandle。每个任务都有一个全局队列，可用于生成/使用输入。

在您的情况下，您的外部程序会将每个目录添加到 Task1 的队列中，并为 Task1 设置 EventWaitHandler。任务 1 将从其 EventWaitHandler 中“唤醒”，获取其队列中的目录计数，然后当计数大于 0 时，从队列中获取目录，扫描所有 .jpg，并添加每个 .jpg 位置到第二个队列，并为任务 2 设置 EventWaitHandle。任务 2 读取其输入，处理它，将其转发到任务 3 的队列...

让所有锁定正常工作可能有点痛苦（我基本上锁定对队列的任何访问，即使是像获取它的计数这样简单的事情）。 .NET 4.0 应该具有自动支持无锁生产者/消费者队列的数据结构。

Answer 3

有趣的问题。 我想出了两种方法。第一个基于PLinq，第二个基于te Rx Framework。

第一个并行遍历文件。 第二个从目录异步生成文件。

这是一个非常简化的版本（第一种方法确实需要 .Net 4.0，因为它使用 PLinq）

string direcory = "Mydirectory";
    var jpegFiles = System.IO.Directory.EnumerateFiles(direcory,"*.jpg");


    // --  PLinq --------------------------------------------
    jpegFiles
    .AsParallel()
    .Select(imageFile => new {OldLocation = imageFile, NewLocation = GenerateCopyLocation(imageFile) })
    .Do(fileInfo => 
        {
            if (!File.Exists(fileInfo.NewLocation ) || 
                (File.GetCreationTime(fileInfo.NewLocation)) != (File.GetCreationTime(fileInfo.NewLocation)))
                File.Copy(fileInfo.OldLocation,fileInfo.NewLocation);
        })
    .Run();

    // -----------------------------------------------------


    //-- Rx Framework ---------------------------------------------
    var resetEvent = new AutoResetEvent(false);
    var doTheWork =
    jpegFiles.ToObservable()
    .Select(imageFile => new {OldLocation = imageFile, NewLocation = GenerateCopyLocation(imageFile) })
    .Subscribe( fileInfo => 
        {
            if (!File.Exists(fileInfo.NewLocation ) || 
                (File.GetCreationTime(fileInfo.NewLocation)) != (File.GetCreationTime(fileInfo.NewLocation)))
            File.Copy(fileInfo.OldLocation,fileInfo.NewLocation);
        },() => resetEvent.Set());

    resetEvent.WaitOne();
    doTheWork.Dispose();

    // -----------------------------------------------------