在 C# 中发送/接收 GZip 压缩的 MSMQ 消息

Question

我正在尝试将大型对象 (>30MB) 发送到 MSMQ 队列。由于我们要发送的数据量很大，因此我们的想法是在发送对象之前先对对象进行 GZip，然后在接收端解压缩。

但是，将压缩流写入message.BodyStream 属性似乎可行，但不能从那里读取。 不知道怎么回事。

Message l_QueueMessage = new Message();
l_QueueMessage.Priority = priority;

using (MessageQueue l_Queue = CreateQueue())
{
    GZipStream stream = new GZipStream(l_QueueMessage.BodyStream, CompressionMode.Compress);

    Formatter.Serialize(stream, message);

    l_Queue.Send(l_QueueMessage);
}

Formatter 是 BinaryFormatter 类型的全局属性。这用于序列化/反序列化为我们想要发送/接收的对象类型，例如“产品项目”。

接收端是这样的：

GZipStream stream = new GZipStream(l_Message.BodyStream, CompressionMode.Decompress);

object decompressedObject = Formatter.Deserialize(stream);

ProductItem l_Item = decompressedObject as ProductItem;

m_ProductReceived(sender, new MessageReceivedEventArgs<ProductItem>(l_Item));

l_ProductQueue.BeginReceive();

我收到一个 EndOfStreamException "{"Unable to read beyond the end of the stream."} 试图反序列化 在 System.IO.BinaryReader.ReadByte()

使用 messageBodyStream 属性，我实际上绕过了 message.Formatter，我没有初始化任何东西，因为我在 GZipStream 中使用了我自己的 ser/deser 机制。但是，我不确定这是否是正确的做法。

我错过了什么？ 谢谢！

Answer 1

在您的原始代码中，问题是您需要关闭GZipStream 才能正确编写 GZip 页脚，然后才能发送它。如果你不这样做，你最终会发送无法反序列化的字节。这也是为什么稍后发送的新代码可以工作的原因。

Answer 2

好的，我完成了这项工作。关键是将接收器上的解压缩流转换为 byte[] 数组。然后反序列化开始工作。

发送方代码（注意发送消息前已关闭流）：

using (MessageQueue l_Queue = CreateQueue())
{
    using (GZipStream stream = new GZipStream(l_QueueMessage.BodyStream, CompressionMode.Compress, true))
    {
        Formatter.Serialize(stream, message);
    }

    l_Queue.Send(l_QueueMessage);
}

接收端（注意我如何将流转换为 byte[] 然后反序列化）：

using (GZipStream stream = new GZipStream(l_QueueMessage.BodyStream, CompressionMode.Decompress))
{
    byte[] bytes = ReadFully(stream);

    using (MemoryStream ms = new MemoryStream(bytes))
    {
        decompressedObject = Formatter.Deserialize(ms);
    }
}

仍然不知道为什么使用 ReadFully() 函数而不是 Stream.CopyTo() 可以工作。 有人吗？

顺便说一句，ReadFully() 是一个从 Stream 中创建 byte[] 的函数。我必须在http://www.yoda.arachsys.com/csharp/readbinary.html 上感谢 Jon Skeet。谢谢！

Answer 3

尽量分开压缩和发送：

byte[] binaryBuffer = null;
using (MemoryStream compressedBody = new MemoryStream()) 
{
    using(GZipStream stream = new GZipStream(compressedBody, CompressionMode.Compress))
    {
        Formatter.Serialize(compressedBody, message); 
        binaryBuffer = compressedBody.GetBuffer();
    }
}

using (MessageQueue l_Queue = CreateQueue())        
{            
    l_QueueMessage.BodyStream.Write(binaryBuffer, 0, binaryBuffer.Length);
    l_QueueMessage.BodyStream.Seek(0, SeekOrigin.Begin);
    l_Queue.Send(l_QueueMessage);
}