c# - 添加 AsParallel() 调用导致我的代码在写入文件时中断

Question

我正在构建一个必须处理一堆文档的控制台应用程序。

为了简单起见，过程是:

1. 对于 X 和 Y 之间的每一年，查询数据库以获取流程的文档引用列表
2. 对于每个引用，处理一个本地文件

我认为 process 方法是独立的，只要输入参数不同就应该并行化:

private static bool ProcessDocument(
        DocumentsDataset.DocumentsRow d,
        string langCode
)
{         
        try
        {                           
            var htmFileName = d.UniqueDocRef.Trim() + langCode + ".htm";
            var htmFullPath = Path.Combine("x:\path", htmFileName;

            missingHtmlFile = !File.Exists(htmFullPath);

            if (!missingHtmlFile)
            {
                var html = File.ReadAllText(htmFullPath);

                // ProcessHtml is quite long : it use a regex search for a list of reference
                // which are other documents, then sends the result to a custom WS

                ProcessHtml(ref html);

                File.WriteAllText(htmFullPath, html);

            }

            return true;             
        }
        catch (Exception exc)
        {
            Trace.TraceError("{0,8}Fail processing {1} : {2}","[FATAL]", d.UniqueDocRef, exc.ToString());
            return false;
        }
    }

为了枚举我的文档，我有这个方法:

    private static IEnumerable<DocumentsDataset.DocumentsRow> EnumerateDocuments()
    {           
        return Enumerable.Range(1990, 2020 - 1990).AsParallel().SelectMany(year => {
            return Document.FindAll((short)year).Documents;
        });
    }

Document 是一个包装文档检索的业务类。此方法的输出是类型化数据集(我正在返回文档表)。该方法等待一年，我确定文档不会超过一年返回(年份实际上是 key 的一部分)。

注意这里使用了 AsParallel()，但我从来没有遇到过这个问题。

现在，我的主要方法是:

        var documents = EnumerateDocuments();

        var result = documents.Select(d => {
            bool success = true;
            foreach (var langCode in new string[] { "-e","-f" })
            {
                success &= ProcessDocument(d, langCode);
            }
            return new { 
                d.UniqueDocRef,
                success
            };
        });
        using (var sw = File.CreateText("summary.csv"))
        {
            sw.WriteLine("Level;UniqueDocRef");
            foreach (var item in result)
            {
                string level;
                if (!item.success) level = "[ERROR]";
                else level = "[OK]";

                sw.WriteLine(
                    "{0};{1}",
                    level,
                    item.UniqueDocRef
                    );
                //sw.WriteLine(item);
            }
        }

此方法在此表单下按预期工作。但是，如果我更换

        var documents = EnumerateDocuments();

通过

        var documents = EnumerateDocuments().AsParrallel();

它停止工作，我不明白为什么。

错误恰好出现在这里(在我的处理方法中):

File.WriteAllText(htmFullPath, html);

它告诉我该文件已被另一个程序打开。

我不明白什么会导致我的程序无法按预期运行。由于我的 documents 变量是一个返回唯一值的 IEnumerable，为什么我的 process 方法出错了？

谢谢你的建议

[编辑] 检索文档的代码:

    /// <summary>
    /// Get all documents in data store
    /// </summary>
    public static DocumentsDS FindAll(short? year)
    {
        Database db = DatabaseFactory.CreateDatabase(connStringName); // MS Entlib

        DbCommand cm = db.GetStoredProcCommand("Document_Select");
        if (year.HasValue) db.AddInParameter(cm, "Year", DbType.Int16, year.Value);

        string[] tableNames = { "Documents", "Years" };
        DocumentsDS ds = new DocumentsDS();
        db.LoadDataSet(cm, ds, tableNames);

        return ds;
    }

[Edit2] 我的问题的可能来源，感谢 mquander。如果我写:

        var test = EnumerateDocuments().AsParallel().Select(d => d.UniqueDocRef);

        var testGr = test.GroupBy(d => d).Select(d => new { d.Key, Count = d.Count() }).Where(c=>c.Count>1);

        var testLst = testGr.ToList();

        Console.WriteLine(testLst.Where(x => x.Count == 1).Count());
        Console.WriteLine(testLst.Where(x => x.Count > 1).Count());

我得到这个结果:

0
1758

删除 AsParallel 返回相同的输出。

结论:我的 EnumerateDocuments 有问题，每个文档返回两次。

我想必须在这里潜水

这可能是我源枚举的原因

Answer 1

我建议您让每个任务将文件数据放入一个全局队列中，并让一个并行线程从队列中获取写入请求并进行实际写入。

无论如何，在单个磁盘上并行写入的性能比顺序写入要差得多，因为磁盘需要旋转以寻找下一个写入位置，所以你只是在寻找之间来回移动磁盘。最好按顺序进行写入。