在C#中的并行编程

Question

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

换句话说，该程序包括四个操作：

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完成后，步骤2就开始了。另一个模型（我认为更有趣，因为我不知道如何做）是创建一种工作者和消费者模型，这样当一个线程完成步骤1时，另一个线程接管并在该对象上执行步骤2（或类似于此）。但是，我不知道这些解决方案是否都可行。而且，我对并行编程并不是很了解。我知道如何创建线程，以及如何让它执行一个以对象为唯一参数的函数，并且我也曾经使用过BackgroundWorker类，但我对它们并不是那么熟悉。

欢迎提供任何意见。

- Alxandr

---

听起来是一个有趣的任务，但由于它很可能会受到输入/输出限制，使用多个线程同时操作磁盘很可能会使程序运行得比只使用一个线程还要_慢_。请注意保持原意不变，并尽可能通俗易懂。 - John Knoeller

谢谢，我没有真正考虑过这个。但是我认为至少步骤2和3可以从使用多个线程中受益，你同意吗？ - Alxandr

http://messagingbus.codeplex.com/ 可能会有所帮助。 - bohdan_trotsenko

Answer 1

以下是几个选项:

• 并行LINQ: 在多核处理器上运行查询

• 任务并行库 (TPL): 为多核机器优化托管代码

• 如果您对基本线程原语和概念感兴趣: C#中的线程

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

- Mitch Wheat

Answer 2

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

单个PDF文件：http://www.albahari.com/threading/threading.pdf。

关于您的第二种方法：

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

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

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

- chocojosh

Answer 3

有趣的问题。 我想到了两种方法。第一种基于PLinq，第二种基于Rx框架。

第一种方法并行迭代文件。 第二种方法异步生成目录中的文件。

以下是一个简化版本的示例（第一种方法需要 .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();

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

- Cheick

---

+1表示提及并提供“Rx”方法的示例。 - Erick Brown