如何在 C# 中保存并附加到序列化的 MessagePack 二进制文件？

Question

我试图用来MessagePack保存多个结构列表，因为我读到它的性能比BinaryFormatter序列化更好。

我想要做的是接收实时时间序列数据并定期将其保存（附加）到磁盘上，例如，如果列表的元素数为 100。我的问题是：

1）在这种情况下，序列化结构列表并将其异步保存到磁盘是否更好？

2) 如何使用 MessagePack 简单地将其保存到磁盘？

public struct struct_realTime
{
    public int indexNum { get; set; }
    public string currentTime { get; set; }
    public string currentType { get; set; }
}

class Program
{
    static void Main(string[] args)
    {
        List<struct_realTime> list_temp = new List<struct_realTime>(100000);

        for (int num=0; num < 100000; num++)
        {
            list_temp.Add(new struct_realTime
            {
                indexNum = 1,
                currentTime = "time",
                currentType = "type",
            });
        }

        string filename = "file.bin";

        using (var fileStream = new FileStream(filename, FileMode.Append, FileAccess.Write))
        {
            byte[] bytes = MessagePackSerializer.Serialize(list_temp);
            Console.WriteLine(MessagePackSerializer.ToJson(bytes));
        }
    }
}

当我运行这段代码时，它创建file.bin并打印出 100000 个结构，但文件是 0 字节。

当我使用时BinaryFormatter，我这样做：

using (var fileStream = new FileStream("file.bin", FileMode.Append))
{
    BinaryFormatter formatter = new BinaryFormatter();
    formatter.Serialize(fileStream, list_temp);
}

我该如何解决这个问题？

Answer 1

您想要做的是将List<struct_realTime>使用序列化的对象（此处）附加MessagePackSerializer到包含类似对象的已序列化序列的文件中，就像使用BinaryFormatter, protobuf-net或Json.NET 一样。稍后，您可能希望能够将整个序列反序列化为相同类型的对象列表或数组。

你的代码有三个问题，两个简单的，一个基本的。

简单的问题如下：

• 您实际上并未写入fileStream. 相反，请执行以下操作：

  // Append each list_temp sequentially
  using (var fileStream = new FileStream(filename, FileMode.OpenOrCreate, FileAccess.ReadWrite))
  {
      MessagePackSerializer.Serialize(fileStream, list_temp);
  }
  

• 您还没有标记struct_realTime与[MessagePackObject]属性。这可以实现，例如如下：

  [MessagePackObject]
  public struct struct_realTime
  {
      [Key(0)]
      public int indexNum { get; set; }
      [Key(1)]
      public string currentTime { get; set; }
      [Key(2)]
      public string currentType { get; set; }
  }
  

完成后，您现在可以重复序列化list_temp到一个文件……但之后您将无法读取它们！那是因为在反序列化根对象时MessagePackSerializer似乎读取整个文件，跳过附加在文件中的任何其他数据。因此，像下面这样的代码将失败，因为只有一个对象从文件中读取：

List<List<struct_realTime>> allItemsInFile = new List<List<struct_realTime>>();
using (var fileStream = File.OpenRead(filename))
{
    while (fileStream.Position < fileStream.Length)
    {
        allItemsInFile.Add(MessagePackSerializer.Deserialize<List<struct_realTime>>(fileStream));                   
    }
}
Assert.IsTrue(allItemsInFile.Count == expectedNumberOfRootItemsInFile);

演示小提琴 #1在这里。

并且像下面这样的代码将失败，因为流中的（第一个）根对象不是对象数组的数组，而只是一个数组：

List<List<struct_realTime>> allItemsInFile;
using (var fileStream = File.OpenRead(filename))
{
    allItemsInFile = MessagePackSerializer.Deserialize<List<List<struct_realTime>>>(fileStream);
}
Assert.IsTrue(allItemsInFile.Count == expectedNumberOfRootItemsInFile);

演示小提琴 #2在这里。

由于MessagePackSerializer似乎缺乏从流中反序列化多个根对象的能力，您有什么选择？首先，您可以反序列化 a List<List<struct_realTime>>，附加到它，然后将整个内容序列化回文件。据推测，出于性能原因，您不想这样做。

其次，直接使用MessagePack 规范，您可以手动查找到文件的开头以解析并重写适当的array 32格式 header，然后查找到文件的末尾并用于MessagePackSerializer序列化和附加新项。以下扩展方法可以完成这项工作：

public static class MessagePackExtensions
{
    const byte Array32 = 0xdd;
    const int Array32HeaderLength = 5;

    public static void AppendToFile<T>(Stream stream, T item)
    {
        if (stream == null)
            throw new ArgumentNullException(nameof(stream));
        if (!stream.CanSeek)
            throw new ArgumentException("!stream.CanSeek");

        stream.Position = 0;
        var buffer = new byte[Array32HeaderLength];
        var read = stream.Read(buffer, 0, Array32HeaderLength);
        stream.Position = 0;
        if (read == 0)
        {
            FormatArray32Header(buffer, 1);
            stream.Write(buffer, 0, Array32HeaderLength);
        }
        else
        {
            var count = ParseArray32Header(buffer, read);
            FormatArray32Header(buffer, count + 1);
            stream.Write(buffer, 0, Array32HeaderLength);
        }

        stream.Position = stream.Length;
        MessagePackSerializer.Serialize(stream, item);
    }

    static void FormatArray32Header(byte [] buffer, uint value)
    {
        buffer[0] = Array32;
        buffer[1] = unchecked((byte)(value >> 24));
        buffer[2] = unchecked((byte)(value >> 16));
        buffer[3] = unchecked((byte)(value >> 8));
        buffer[4] = unchecked((byte)value);
    }

    static uint ParseArray32Header(byte [] buffer, int readCount)
    {
        if (readCount < 5 || buffer[0] != Array32)
            throw new ArgumentException("Stream was not positioned on an Array32 header.");
        int i = 1;
        ///sf/ask/576874231/
        ///sf/answers/576878921/ by /sf/users/1634811/
        var value = unchecked((uint)((buffer[i++] << 24) | (buffer[i++] << 16) | (buffer[i++] << 8) | buffer[i++]));
        return value;
    }
}

它可用于附加您list_temp的如下内容：

// Append each entry sequentially
using (var fileStream = new FileStream(filename, FileMode.OpenOrCreate, FileAccess.ReadWrite))
{
    MessagePackExtensions.AppendToFile(fileStream, list_temp);
}

然后，要反序列化整个文件，请执行以下操作：

List<List<struct_realTime>> allItemsInFile;
using (var fileStream = File.OpenRead(filename))
{
    allItemsInFile = MessagePackSerializer.Deserialize<List<List<struct_realTime>>>(fileStream);
}

笔记：

• MessagePack 协议有 3 种不同的数组格式：

  • fixarray 存储长度最多为 15 个元素的数组。
  • array 16 存储长度最多为 (2^16)-1 个元素的数组。
  • array 32 存储长度最多为 (2^32)-1 个元素的数组。

  
  扩展方法要求根数组array 32消除当新大小变得大于fixarrayor的容量时重新格式化整个数组的需要array 16。 MessagePackSerializer，但是，将始终写入最紧凑的格式，因此MessagePackSerializer不能保证附加到先前序列化的集合中。

• 如果您想使用不需要文件开头的数组计数或大小的快速二进制序列化程序，从而支持开箱即用的追加操作，请考虑protobuf-net。有关详细信息，请参阅我有一个文件并且需要随机序列化多个对象。我如何在 c# 中？以及如何在使用 c# protobuf-net 进行序列化时将对象附加到文件？.

  有关如何使用此序列化程序的一般概述，请参阅https://github.com/protobuf-net/protobuf-net#protobuf-net和Protobuf-net：非官方手册。您需要使用类似于MessagePackSerializer.

演示小提琴 #3在这里。