Разобрать огромный OData JSON путем потоковой передачи определенных разделов JSON, чтобы избежать LOH - PullRequest
Новая
       44

Разобрать огромный OData JSON путем потоковой передачи определенных разделов JSON, чтобы избежать LOH

3 голосов
/ 08 мая 2019

У меня есть ответ OData в виде JSON (, что составляет несколько МБ ) , и требуется потоковая передача "определенных частей JSON", даже не загружая их в память.

Например : когда я читаю свойство "value[0].Body.Content" в следующем JSON (которое будет в МБ), я хочу передать эту часть значениябез десериализации его в объект типа строка.Поэтому в основном считайте часть значения в байтовый массив фиксированного размера и запишите этот байтовый массив в целевой поток (повторяя шаг до тех пор, пока эти данные не закончат обработку).

JSON: 

{
    "@odata.context": "https://localhost:5555/api/v2.0/$metadata#Me/Messages",
    "value": [
        {
            "@odata.id": "https://localhost:5555/api/v2.0/",
            "@odata.etag": "W/\"Something\"",
            "Id": "vccvJHDSFds43hwy98fh",
            "CreatedDateTime": "2018-12-01T01:47:53Z",
            "LastModifiedDateTime": "2018-12-01T01:47:53Z",
            "ChangeKey": "SDgf43tsdf",
            "WebLink": "https://localhost:5555/?ItemID=dfsgsdfg9876ijhrf",
            "Body": {
                "ContentType": "HTML",
                "Content": "<html>\r\n<body>Huge Data Here\r\n</body>\r\n</html>\r\n"
            },
            "ToRecipients": [{
                    "EmailAddress": {
                        "Name": "ME",
                        "Address": "me@me.com"
                    }
                }
            ],
            "CcRecipients": [],
            "BccRecipients": [],
            "ReplyTo": [],
            "Flag": {
                "FlagStatus": "NotFlagged"
            }
        }
    ],
    "@odata.nextLink": "http://localhost:5555/rest/jersey/sleep?%24filter=LastDeliveredDateTime+ge+2018-12-01+and+LastDeliveredDateTime+lt+2018-12-02&%24top=50&%24skip=50"
}

Подходы пробовали: 1. Newtonsoft

Сначала я попытался использовать потоковую передачу Newtonsoft, но она внутренне преобразует данные в строку и загружает в память .(Это приводит к тому, что LOH начинает работать, а память не освобождается до тех пор, пока не произойдет сжатие - у нас есть ограничение памяти для нашего рабочего процесса, и мы не можем сохранить его в памяти)

**code:**

    using (var jsonTextReader = new JsonTextReader(sr))
    {
        var pool = new CustomArrayPool();
        // Checking if pooling will help with memory
        jsonTextReader.ArrayPool = pool;

        while (jsonTextReader.Read())
        {
            if (jsonTextReader.TokenType == JsonToken.PropertyName
                && ((string)jsonTextReader.Value).Equals("value"))
            {
                jsonTextReader.Read();

                if (jsonTextReader.TokenType == JsonToken.StartArray)
                {
                    while (jsonTextReader.Read())
                    {
                        if (jsonTextReader.TokenType == JsonToken.StartObject)
                        {
                            var Current = JToken.Load(jsonTextReader);
                            // By Now, the LOH Shoots up.
                            // Avoid below code of converting this JToken back to byte array.
                            destinationStream.write(Encoding.ASCII.GetBytes(Current.ToString()));
                        }
                        else if (jsonTextReader.TokenType == JsonToken.EndArray)
                        {
                            break;
                        }
                    }
                }
            }

            if (jsonTextReader.TokenType == JsonToken.StartObject)
            {
                var Current = JToken.Load(jsonTextReader);
                // Do some processing with Current
                destinationStream.write(Encoding.ASCII.GetBytes(Current.ToString()));
            }
        }
    }

OData.Net:

Я думал, можно ли это сделать с помощью библиотеки OData.Net, поскольку она выглядит так, как будто она поддерживает потоковую передачу строковых полей ,Но я не мог уйти далеко, так как в итоге я создал Модель для данных, что означало бы, что значение будет преобразовано в один строковый объект из МБ.

Код 

ODataMessageReaderSettings settings = new ODataMessageReaderSettings();
IODataResponseMessage responseMessage = new InMemoryMessage { Stream = stream };
responseMessage.SetHeader("Content-Type", "application/json;odata.metadata=minimal;");
// ODataMessageReader reader = new ODataMessageReader((IODataResponseMessage)message, settings, GetEdmModel());
ODataMessageReader reader = new ODataMessageReader(responseMessage, settings, new EdmModel());
var oDataResourceReader = reader.CreateODataResourceReader();
var property = reader.ReadProperty();

Есть идеи, как разбирать этот JSON по частям, используя OData.Net/Newtonsoft и потоковое значение определенных полей?Это единственный способ сделать это, вручную проанализировать поток?

1 Ответ

3 голосов
/ 09 мая 2019

Если вы копируете части JSON из одного потока в другой, вы можете сделать это более эффективно с помощью JsonWriter.WriteToken(JsonReader), таким образом избегая промежуточных представлений Current = JToken.Load(jsonTextReader) и Encoding.ASCII.GetBytes(Current.ToString()) и связанных с ними накладных расходов памяти: 

using (var textWriter = new StreamWriter(destinationStream, new UTF8Encoding(false, true), 1024, true))
using (var jsonWriter = new JsonTextWriter(textWriter) { Formatting = Formatting.Indented, CloseOutput = false })
{
    // Use Formatting.Indented or Formatting.None as required.
    jsonWriter.WriteToken(jsonTextReader);
}

Однако, Json.NET JsonTextReader не может читать одно строковое значение в «чанках» так же, как XmlReader.ReadValueChunk(). Он всегда полностью материализует каждое атомарное значение строки. Если значения строк настолько велики, что они собираются в куче больших объектов, даже использование JsonWriter.WriteToken() не помешает полной загрузке этих строк в память.

В качестве альтернативы вы можете рассмотреть читателей и писателей, возвращенных JsonReaderWriterFactory. Эти читатели и писатели используются DataContractJsonSerializer и переводят JSON в XML на лету в том виде, в каком они читаются и пишутся . Поскольку базовыми классами для этих читателей и писателей являются XmlReader и XmlWriter, они do поддерживают чтение и запись строковых значений в чанках. Правильное их использование позволит избежать размещения строк в куче больших объектов.

Для этого сначала определите следующие методы расширения, которые копируют выбранное подмножество значений (значений) JSON из входного потока в выходной поток, как указано путем к данным для потоковой передачи: 

public static class JsonExtensions
{
    public static void StreamNested(Stream from, Stream to, string [] path)
    {
        var reversed = path.Reverse().ToArray();

        using (var xr = JsonReaderWriterFactory.CreateJsonReader(from, XmlDictionaryReaderQuotas.Max))
        {
            foreach (var subReader in xr.ReadSubtrees(s => s.Select(n => n.LocalName).SequenceEqual(reversed)))
            {
                using (var xw = JsonReaderWriterFactory.CreateJsonWriter(to, Encoding.UTF8, false))
                {
                    subReader.MoveToContent();

                    xw.WriteStartElement("root");
                    xw.WriteAttributes(subReader, true);

                    subReader.Read();

                    while (!subReader.EOF)
                    {
                        if (subReader.NodeType == XmlNodeType.Element && subReader.Depth == 1)
                            xw.WriteNode(subReader, true);
                        else
                            subReader.Read();
                    }

                    xw.WriteEndElement();
                }
            }
        }
    }
}

public static class XmlReaderExtensions
{
    public static IEnumerable<XmlReader> ReadSubtrees(this XmlReader xmlReader, Predicate<Stack<XName>> filter)
    {
        Stack<XName> names = new Stack<XName>();

        while (xmlReader.Read())
        {
            if (xmlReader.NodeType == XmlNodeType.Element)
            {
                names.Push(XName.Get(xmlReader.LocalName, xmlReader.NamespaceURI));
                if (filter(names))
                {
                    using (var subReader = xmlReader.ReadSubtree())
                    {
                        yield return subReader;
                    }
                }
            }

            if ((xmlReader.NodeType == XmlNodeType.Element && xmlReader.IsEmptyElement)
                || xmlReader.NodeType == XmlNodeType.EndElement)
            {
                names.Pop();
            }
        }
    }
}

Теперь аргумент string [] path для StreamNested() - это , а не любой путь . Вместо этого это путь, соответствующий иерархии элементов XML, соответствующих JSON, который вы хотите выбрать в переводе XmlReader, возвращаемом JsonReaderWriterFactory.CreateJsonReader(). Отображение, используемое для этот перевод, в свою очередь, задокументирован Microsoft в Отображение между JSON и XML . Для выбора и потоковой передачи только тех значений JSON, которые соответствуют value[*], требуется XML-путь //root/value/item. Таким образом, вы можете выбирать и передавать нужные вложенные объекты, выполнив: 

JsonExtensions.StreamNested(inputStream, destinationStream, new[] { "root", "value", "item" });

Примечания:

  • Отображение между JSON и XML довольно сложно. Часто проще просто загрузить образец JSON в XDocument, используя следующий метод расширения: 

    static XDocument ParseJsonAsXDocument(string json)
{
    using (var xr = JsonReaderWriterFactory.CreateJsonReader(new MemoryStream(Encoding.UTF8.GetBytes(json)), Encoding.UTF8, XmlDictionaryReaderQuotas.Max, null))
    {
        return XDocument.Load(xr);
    }
}

    А затем определить правильный XML-путь на основе наблюдений.

  • По вопросам, связанным с, см. Эквивалент JObject.SelectToken в .NET .

...