c# - protobuf-net 使用 SerializeWithLengthPrefix 序列化对象的嵌套列表

Question

我目前正在尝试使用 protobuf-net 序列化以下数据结构:

[ProtoContract]
public class Recording
{
    [ProtoMember(1)]
    public string Name;

    [ProtoMember(2)]
    public List<Channel> Channels;
}

[ProtoContract]
public class Channel
{
    [ProtoMember(1)]
    public string ChannelName;

    [ProtoMember(2)]
    public List<float> DataPoints;
}

我有固定数量的 12 个 channel ，但是每个 channel 的数据点数量可能会变得非常大(所有 channel 的数据点数量可达 Gb 范围)。 因此(并且因为数据是连续流)，我不想一次读取和保存一个记录的结构，而是利用 SerializeWithLengthPrefix (和 DeserializeItems)连续保存它。 我的问题是，是否可以使用这样的嵌套结构来做到这一点，或者我是否必须将其展平？ 我已经看到了第一层次结构级别中的列表示例，但没有看到适合我的具体情况的示例。 另外，如果我将数据点写为 10、100、... 的“ block ”(例如使用列表而不是列表)比直接序列化它们有什么好处吗？

预先感谢您的帮助

托比亚斯

Answer 1

您尝试做的事情的关键挑战是每个对象的内部都严重基于流。 protobuf-net 可以以这种方式工作，但这并不简单。还有一个问题是您希望将来自单个 channel 的数据交错到多个片段上，这不是惯用的 protobuf 布局。因此，核心对象物化器代码可能无法完全满足您的要求 - 即，将其视为开放流，而不是全部加载到内存中，以进行读取和写入。

也就是说:您可以使用原始读取器/写入器 API 来实现流式传输。您可能应该使用 BinaryWriter/BinaryReader 与类似代码进行比较和对比，但本质上是以下工作:

using ProtoBuf;
using System;
using System.Collections.Generic;
using System.IO;
static class Program
{
    static void Main()
    {
        var path = "big.blob";
        WriteFile(path);

        int channelTotal = 0, pointTotal = 0;
        foreach(var channel in ReadChannels(path))
        {
            channelTotal++;
            pointTotal += channel.Points.Count;
        }
        Console.WriteLine("Read: {0} points in {1} channels", pointTotal, channelTotal);
    }
    private static void WriteFile(string path)
    {
        string[] channels = {"up", "down", "top", "bottom", "charm", "strange"};
        var rand = new Random(123456);

        int totalPoints = 0, totalChannels = 0;
        using (var encoder = new DataEncoder(path, "My file"))
        {
            for (int i = 0; i < 100; i++)
            {
                var channel = new Channel {
                    Name = channels[rand.Next(channels.Length)]
                };
                int count = rand.Next(1, 50);
                var data = new List<float>(count);
                for (int j = 0; j < count; j++)
                    data.Add((float)rand.NextDouble());
                channel.Points = data;
                encoder.AddChannel(channel);
                totalPoints += count;
                totalChannels++;
            }
        }

        Console.WriteLine("Wrote: {0} points in {1} channels; {2} bytes", totalPoints, totalChannels, new FileInfo(path).Length);
    }
    public class Channel
    {
        public string Name { get; set; }
        public List<float> Points { get; set; }
    }
    public class DataEncoder : IDisposable
    {
        private Stream stream;
        private ProtoWriter writer;
        public DataEncoder(string path, string recordingName)
        {
            stream = File.Create(path);
            writer = new ProtoWriter(stream, null, null);

            if (recordingName != null)
            {
                ProtoWriter.WriteFieldHeader(1, WireType.String, writer);
                ProtoWriter.WriteString(recordingName, writer);
            }
        }
        public void AddChannel(Channel channel)
        {
            ProtoWriter.WriteFieldHeader(2, WireType.StartGroup, writer);
            var channelTok = ProtoWriter.StartSubItem(null, writer);

            if (channel.Name != null)
            {
                ProtoWriter.WriteFieldHeader(1, WireType.String, writer);
                ProtoWriter.WriteString(channel.Name, writer);
            }
            var list = channel.Points;
            if (list != null)
            {

                switch(list.Count)
                {
                    case 0:
                        // nothing to write
                        break;
                    case 1:
                        ProtoWriter.WriteFieldHeader(2, WireType.Fixed32, writer);
                        ProtoWriter.WriteSingle(list[0], writer);
                        break;
                    default:
                        ProtoWriter.WriteFieldHeader(2, WireType.String, writer);
                        var dataToken = ProtoWriter.StartSubItem(null, writer);
                        ProtoWriter.SetPackedField(2, writer);
                        foreach (var val in list)
                        {
                            ProtoWriter.WriteFieldHeader(2, WireType.Fixed32, writer);
                            ProtoWriter.WriteSingle(val, writer);
                        }
                        ProtoWriter.EndSubItem(dataToken, writer);
                        break;
                }
            }
            ProtoWriter.EndSubItem(channelTok, writer);
        }
        public void Dispose()
        {
            using (writer) { if (writer != null) writer.Close(); }
            writer = null;
            using (stream) { if (stream != null) stream.Close(); }
            stream = null;
        }
    }

    private static IEnumerable<Channel> ReadChannels(string path)
    {
        using (var file = File.OpenRead(path))
        using (var reader = new ProtoReader(file, null, null))
        {
            while (reader.ReadFieldHeader() > 0)
            {
                switch (reader.FieldNumber)
                {
                    case 1:
                        Console.WriteLine("Recording name: {0}", reader.ReadString());
                        break;
                    case 2: // each "2" instance represents a different "Channel" or a channel switch
                        var channelToken = ProtoReader.StartSubItem(reader);
                        int floatCount = 0;
                        List<float> list = new List<float>();
                        Channel channel = new Channel { Points = list };
                        while (reader.ReadFieldHeader() > 0)
                        {

                            switch (reader.FieldNumber)
                            {
                                case 1:
                                    channel.Name = reader.ReadString();
                                    break;
                                case 2:
                                    switch (reader.WireType)
                                    {
                                        case WireType.String: // packed array - multiple floats
                                            var dataToken = ProtoReader.StartSubItem(reader);
                                            while (ProtoReader.HasSubValue(WireType.Fixed32, reader))
                                            {
                                                list.Add(reader.ReadSingle());
                                                floatCount++;
                                            }
                                            ProtoReader.EndSubItem(dataToken, reader);
                                            break;
                                        case WireType.Fixed32: // simple float
                                            list.Add(reader.ReadSingle());
                                            floatCount++; // got 1
                                            break;
                                        default:
                                            Console.WriteLine("Unexpected data wire-type: {0}", reader.WireType);
                                            break;
                                    }
                                    break;
                                default:
                                    Console.WriteLine("Unexpected field in channel: {0}/{1}", reader.FieldNumber, reader.WireType);
                                    reader.SkipField();
                                    break;
                            }
                        }
                        ProtoReader.EndSubItem(channelToken, reader);
                        yield return channel;
                        break;
                    default:
                        Console.WriteLine("Unexpected field in recording: {0}/{1}", reader.FieldNumber, reader.WireType);
                        reader.SkipField();
                        break;
                }
            }
        }
    }
}