反序列化对象数组的二进制数据作为元素可用

Question

我一直在筛选帖子和论坛，但找不到实现这一目标的方法。

我有一个包含 10,000,000 个 Person 对象的数组。我正在使用 Streamed WCF Net.Tcp Web 服务通过网络发送这些对象。

问题是我想读取数组的第一个，例如 5000 个 Person 对象，因为它到达并只处理那些。之后，我将推进流并阅读另外 5000 条，等等......

我无法找到一种方法来做到这一点，因为据我所知，C# 中没有明确的对象大小。例如，我不能只读取流的前 312 字节并说“是的，这是第一个 Person 对象。现在读取接下来的 312 字节以获取下一个人。”。

理想情况下，我希望使用 ProtoBuf-Net 来序列化我的对象，但 .NET BinaryFormatter 也可以。

我也愿意以块的形式发送数据，例如 5000 的数组。但我不想每次都打开全新的 tcp 连接。如果只有一种方法可以告诉读取流的代码：“好的，反序列化我刚刚发送给您的所有内容（5000 个数组），然后我将继续将另一个 5000 写入流”。

有什么想法吗？ 谢谢。

Answer 1

.NET 中的大多数对象可能没有明确的大小，但您可以找到序列化对象的大小。首先发送序列化对象的大小（以字节为单位），然后发送序列化对象。

// psuedo-code
byte[] serializedObj = DoSerialization(Person);    // we see length on an array

using (var writer = new StreamWriter(stream)) {
    writer.Write(serializedObj.Length);
    stream.Write(serializedObj);
}

您还可以通过修改发送对象的内容和方式来批量执行此操作。你可以创建一个List<Person>，添加N个Person，序列化List并像以前一样发送。

虽然我不确定是否有必要在发送数据之前发送大小，但它可以帮助您在读取流时了解您期望的字节数。

Answer 2

您可以使用 protobuf-net 执行此操作，只需在接收系统中使用 ObservableCollection<Person>。当反序列化期间集合增长到大于 5000 个对象时，请在 ObservableCollection<T>.CollectionChanged 回调中删除并处理这些项目。然后在 [OnDeserialized] 回调中处理任何剩余的项目。

例如，考虑以下根对象：

[ProtoContract]
public class RootObject
{
    public RootObject()
    {
        this.People = new ObservableCollection<Person>();
    }

    [ProtoMember(1)]
    public ObservableCollection<Person> People { get; private set; }

    public event EventHandler<EventArgs<StreamingContext>> OnDeserialized;

    [OnDeserialized]
    internal void OnDeserializedMethod(StreamingContext context)
    {
        var onDeserialized = OnDeserialized;
        if (onDeserialized != null)
            onDeserialized(this, new EventArgs<StreamingContext> { Value = context });
    }
}

public class EventArgs<T> : EventArgs
{
    public T Value { get; set; }
}

假设您想调用一个方法来处理每个被添加到集合中的 5000 个Person 对象，例如：

const int ProcessIncrement = 5000;

void ProcessItems(ICollection<Person> people, bool force)
{
    if (people == null || people.Count == 0)
        return;
    if (people.Count >= ProcessIncrement || force)
    {
        // Remove and process the items, possibly on a different thread.
        Console.WriteLine(string.Format("Processing {0} people." people.Count));
        people.Clear();
    }
}

您可以预先分配您的RootObject 并添加具有必要逻辑的侦听器，并将序列化流的内容合并到根：

// Allocate a new RootObject
var newRoot = new RootObject();

// Add listeners to process chunks of Person objects as they are added
newRoot.People.CollectionChanged += (o, e) =>
    {
        // Process each chunk of 5000.
        var collection = (ICollection<Person>)o;
        ProcessItems(collection, false);
    };

newRoot.OnDeserialized += (o, e) =>
    {
        // Forcibly process any remaining no matter how many.
        ProcessItems(((RootObject)o).People, true);
    };

// Deserialize from the stream onto the pre-allocated newRoot
Serializer.Merge(stream, newRoot);

根据需要，每次将对象添加到集合时都会调用ProcessItems，以 5000 为增量处理它们，然后无条件处理剩余部分。

现在，唯一的问题是，protobuf-net 是在反序列化集合之前将整个流加载到内存中，还是进行流反序列化？事实证明，它是后者，正如sample fiddle 所示，随着People 集合中的项目被添加、处理和删除，流位置逐渐增加。

这里我在反序列化之前手动将监听器添加到RootObject。如果您要将它们添加到构造函数本身，您可以在预分配的根对象上使用 ProtoBuf.Serializer.Deserialize<RootObject>(Stream stream) 而不是 Serializer.Merge，这可能更容易集成到您当前的架构中。

顺便说一下，这种技术也应该适用于 XmlSerializer 和 Json.NET。