并发 XmlReader 和 XmlWriter答案

【问题标题】：Concurrent XmlReader and XmlWriter并发 XmlReader 和 XmlWriter
【发布时间】：2014-04-24 18:10:00
【问题描述】：

我有一个系统创建一些XmlReader 和XmlWriter 实例，并将它们传递给客户端，以便客户端可以写入和读取 XML。

有时，多个线程访问系统：

线程 1 获得一个 XmlWriter 实例并开始写入
线程 2 获得一个代表相同 XML 文档的 XmlReader 实例，并开始读取。

显然，有时读者在写者完成写入之前就开始阅读，这会导致读者抛出异常：

System.Xml.XmlException : 缺少根元素。

这并不特别令人惊讶，但是有什么方法可以让这个系统线程安全？

是否可以让读者阅读而作者写作？我尝试使用 MemoryStream 作为底层数据存储，但这似乎不起作用。
是否可以让读者等到作者完成写作？
是否可以检测到作者仍在写作？

我相信，以上任何一种方法都可以解决我的问题。写入和读取不需要同时发生。实际上，如果我可以简单地告诉请求XmlReader 的客户读者还没有准备好，我会非常高兴，它必须稍后再试。但是，这需要有一种方法来检测作者仍在写作。

我可以完全控制创建 XmlReader 和 XmlWriter 实例的代码，但我无法更改公共 API。我可以使用string、StringBuilder、MemoryStream 或任何其他可以完成工作的内存中对象作为底层存储，我可以使用XmlReader 和@987654333 的任何派生类@。

有没有办法让这个系统线程安全？

【问题讨论】：

您需要创建一个结帐/签入机制。
您可能会创建自己的流作为 XML 读取器/写入器的提供程序，并且它会阻止读取操作，直到它知道所有写入都已完成。一旦发生写入，取消阻止读取并愉快地继续。您可能仍然可以使用 MemoryStream 作为自定义流的支持。
当您说“客户”时，您指的是多个进程吗？（请告诉我不）。
@PeterRitchie 不，只是同一进程中的其他对象（但在不同的线程上）。
@LB2：流级别的阻塞可能不够，因为流不知道操作何时真正完成。

标签： c# xml multithreading

【解决方案1】：

您最好的选择可能是创建派生自XmlReader 和XmlWriter 的自定义类。您可以向它们添加一个通用的同步对象，并使用ReadWriterLockSlim 之类的东西来协调读写。当你说它“完成”时，你必须在 writer 类中确定你的意思（处置？完成了编写单个节点，即使它没有完全完成？等等）

然后，您可以将读取器中的方法设置为阻塞，直到没有进行写入。

【讨论】：

谢谢，这个解决方案就是我所需要的。事实上，我只需要创建一个派生自 XmlWriter 的类就可以了，其中包括在处理它时的回调。在回调发生之前，我只是告诉读者文档还没有准备好，然后当回调发生时，文档变为可用。

【解决方案2】：

您正在寻找的是一个允许一个线程写入而另一个线程读取的流。 .NET Framework 不提供这样的东西。您可以使用命名管道来执行此操作，但我发现这很麻烦。所以我写了我所谓的ProducerConsumerStream。它只是一个包裹在Stream 接口中的循环队列。有关说明和完整来源，请参阅 Building a new type of stream。

要将它与XmlWriter 和XmlReader 一起使用，您需要执行以下操作：

const int BufferSize = 1024 * 1024;  // megabyte stream buffer

var pcStream = new ProducerConsumerStream(BufferSize);

// start producer thread, passing it the stream
// start consumer thread, passing it the stream

// Destroy the stream when the producer and consumer are done

生产者线程将创建一个写入流的XmlWriter：

using (var writer = XmlWriter.Create(pcStream, writerSettings))
{
    // write xml here
}

// when you're done writing XML, call CompleteAdding on the stream.
// This will let the reader know when the stream is ended.
pcStream.CompleteAdding();

读者线程类似：

using (var reader = XmlReader.Create(pcStream, readerSettings))
{
    // read XML here
}

确保在您的写入器设置中设置了CloseOutput = false，并在读取器设置中设置了CloseInput = false。否则流可能会过早地被释放。

这里的关键是ProducerConsumerStream.Read 阻塞直到它可以读取数据。除非已通过调用CompleteAdding 发出流结束信号。

明白，这是一个内存结构。如果您还想保留 XML，则必须采用其他方式。虽然我想您可以从 ProducerConsumerStream 派生并覆盖 Write 方法，以便它输出到文件以及将内容复制到内部缓冲区。

【讨论】：

这看起来像是解决此类问题的正确通用解决方案（所以 +1）。我希望这已经内置到 BCL 中，而我只是忽略了一些明显的东西，但事实似乎并非如此。但是，由于这个建议的解决方案相当复杂，而且我可以使用更简单的解决方案来管理，所以我选择了另一个答案作为“the”答案。谢谢。

【解决方案3】：

听起来您需要类似于 Java 中的 PipedReader/PipedWriter 的东西。 http://docs.oracle.com/javase/7/docs/api/java/io/PipedReader.html。看起来 .NET 没有类似的东西：Java's PipedReader / PipedWriter equivalents in C#?。

您可以通过使用此 Java 代码作为基础实现您自己的 System.IO.TextWriter 和 System.IO.TextReader 来做与 Java 中相同的事情：

http://grepcode.com/file_/repository.grepcode.com/java/root/jdk/openjdk/7-b147/java/io/PipedReader.java/?v=source

http://grepcode.com/file_/repository.grepcode.com/java/root/jdk/openjdk/7-b147/java/io/PipedWriter.java/?v=source

【讨论】：

这看起来像是解决此类问题的正确通用解决方案（所以 +1）。我希望这已经内置到 BCL 中，而我只是忽略了一些明显的东西，但事实似乎并非如此。但是，由于这个建议的解决方案相当复杂，我可以使用更简单的解决方案来管理，因此我选择了另一个答案作为“the”答案。谢谢。

【解决方案4】：

是否可以让读者等到作者完成写作？

是的，hybrid thread synchronization construct，例如ManualResetEventSlim，如下：

internal class Program
{
    private static void Main()
    {
        var readOperation = new ManualResetEventSlim();
        var stream = new MemoryStream();

        Task.Factory.StartNew(() =>
        {
            using (var writer = XmlWriter.Create(stream))
            {
                Console.WriteLine(
                    "Thread {0} is writing...", 
                    Thread.CurrentThread.ManagedThreadId);
                Thread.Sleep(3000);

                CreateXmlDocument(writer);
            }
            stream.Position = 0;
            readOperation.Set();
        });

        Task.Factory.StartNew(() =>
        {
            readOperation.Wait();
            using (var reader = XmlReader.Create(stream))
                while (reader.Read())
                    Console.WriteLine(
                        "Thread {0} is reading...", 
                        Thread.CurrentThread.ManagedThreadId);
        });

        Console.ReadKey();
    }
}

控制台输出：

Thread 7 is writing...
Thread 11 is reading...
Thread 11 is reading...
Thread 11 is reading...
Thread 11 is reading...
Thread 11 is reading...
Thread 11 is reading...
Thread 11 is reading...
Thread 11 is reading...
Thread 11 is reading...
Thread 11 is reading...
Thread 11 is reading...
Thread 11 is reading...
Thread 11 is reading...
Thread 11 is reading...

CreateXmlDocument 函数（虽然不是很有趣）定义为：

private static void CreateXmlDocument(XmlWriter writer)
{
    writer.WriteStartDocument();
    writer.WriteStartElement("Product");
    writer.WriteAttributeString("ID", "001");
    writer.WriteAttributeString("Name", "Soap");
    writer.WriteElementString("Price", "10.00");
    writer.WriteStartElement("OtherDetails");
    writer.WriteElementString("BrandName", "X Soap");
    writer.WriteElementString("Manufacturer", "X Company");
    writer.WriteEndElement();
    writer.WriteEndDocument();
}

【讨论】：

此解决方案依赖于客户端的同步。但是，我正在寻找一种不需要同步客户端的解决方案，因为作者和读者可能彼此不了解。
哦，我明白了...我强调您可以完全控制创建 XmlReader 和 XmlWriter 实例的代码。

【解决方案5】：

一个简单的想法是为当前正在编写的文档创建一个（可能是静态的）字典。当该文档在该字典中时，读取请求要么被阻止，要么被拒绝。

【讨论】：

【解决方案6】：

好主意是在锁中创建 XML 文件的副本，该锁也会影响 XMLWriter。

编写器代码：

lock(_WriteLock)
     _MyXmlWriter.Flush();

阅读器代码：

lock(_WriteLock)
     File.Copy(myXMLFilePath, tempXMLFilePath);
_MyXMLReader = new XMLReader(tempXMLFilePath);

【讨论】：

锁定文件为什么要复制？