Асинхронная десериализация списка с использованием System.Text.Json

Допустим, я запрашиваю большой файл JSON, который содержит список многих объектов. Я не хочу, чтобы они все время оставались в памяти, но я бы предпочел прочитать и обработать их один за другим. Поэтому мне нужно превратить асинхронный System.IO.Streamпоток в IAsyncEnumerable<T>. Как мне использовать новый System.Text.JsonAPI для этого?

private async IAsyncEnumerable<T> GetList<T>(Uri url, CancellationToken cancellationToken = default)
{
    using (var httpResponse = await httpClient.GetAsync(url, cancellationToken))
    {
        using (var stream = await httpResponse.Content.ReadAsStreamAsync())
        {
            // Probably do something with JsonSerializer.DeserializeAsync here without serializing the entire thing in one go
        }
    }
}

Да, действительно потоковый JSON (de) сериализатор был бы хорошим улучшением производительности во многих местах.

К сожалению, System.Text.Jsonне делает этого в настоящее время. Я не уверен, будет ли это в будущем - я надеюсь на это! По-настоящему потоковая десериализация JSON оказывается довольно сложной задачей.

Вы можете проверить, поддерживает ли чрезвычайно быстрый Utf8Json его, возможно.

Однако может быть индивидуальное решение для вашей конкретной ситуации, поскольку ваши требования, похоже, ограничивают сложность.

Идея состоит в том, чтобы вручную читать один элемент из массива за раз. Мы используем тот факт, что каждый элемент списка сам по себе является допустимым объектом JSON.

Вы можете вручную пропустить [(для первого элемента) или ,(для каждого следующего элемента). Тогда я думаю, что вам лучше всего использовать .NET Core, Utf8JsonReaderчтобы определить, где заканчивается текущий объект, и передать отсканированные байты JsonDeserializer.

Таким образом, вы только слегка буферизуете по одному объекту за раз.

И так как мы говорим о производительности, вы можете получить информацию от a PipeReader, пока вы это делаете. :-)

TL; DR Это не тривиально

Похоже, кто-то уже опубликовал полный код для Utf8JsonStreamReaderструктуры, которая считывает буферы из потока и передает их в Utf8JsonRreader, что позволяет легко десериализовать с помощью JsonSerializer.Deserialize<T>(ref newJsonReader, options);. Код тоже не тривиален. Соответствующий вопрос здесь, а ответ здесь .

Однако этого недостаточно - HttpClient.GetAsyncон вернется только после получения полного ответа, по существу буферизуя все в памяти.

Чтобы избежать этого, следует использовать HttpClient.GetAsync (string, HttpCompletionOption)HttpCompletionOption.ResponseHeadersRead .

Цикл десериализации должен также проверять токен отмены, и либо выходить, либо выбрасывать, если он сигнализирован. В противном случае цикл будет продолжаться, пока весь поток не будет получен и обработан.

Этот код основан на примере соответствующего ответа и использует HttpCompletionOption.ResponseHeadersReadи проверяет токен отмены. Он может анализировать строки JSON, которые содержат правильный массив элементов, например:

[{"prop1":123},{"prop1":234}]

Первый вызов jsonStreamReader.Read()перемещается в начало массива, а второй - в начало первого объекта. Сам цикл завершается, когда обнаружен конец array ( ]).

private async IAsyncEnumerable<T> GetList<T>(Uri url, CancellationToken cancellationToken = default)
{
    //Don't cache the entire response
    using var httpResponse = await httpClient.GetAsync(url,                               
                                                       HttpCompletionOption.ResponseHeadersRead,  
                                                       cancellationToken);
    using var stream = await httpResponse.Content.ReadAsStreamAsync();
    using var jsonStreamReader = new Utf8JsonStreamReader(stream, 32 * 1024);

    jsonStreamReader.Read(); // move to array start
    jsonStreamReader.Read(); // move to start of the object

    while (jsonStreamReader.TokenType != JsonTokenType.EndArray)
    {
        //Gracefully return if cancellation is requested.
        //Could be cancellationToken.ThrowIfCancellationRequested()
        if(cancellationToken.IsCancellationRequested)
        {
            return;
        }

        // deserialize object
        var obj = jsonStreamReader.Deserialize<T>();
        yield return obj;

        // JsonSerializer.Deserialize ends on last token of the object parsed,
        // move to the first token of next object
        jsonStreamReader.Read();
    }
}

JSON-фрагменты, AKA потокового JSON ака ... *

В сценариях потоковой передачи событий или журналирования достаточно часто добавлять отдельные объекты JSON в файл, по одному элементу в строке, например:

{"eventId":1}
{"eventId":2}
...
{"eventId":1234567}

Это не действительный документ JSON, но отдельные фрагменты действительны. Это имеет несколько преимуществ для больших данных / сценариев с высокой степенью одновременности. Добавление нового события требует только добавления новой строки в файл, а не анализа и перекомпоновки всего файла. Обработка , особенно параллельная, проще по двум причинам:

• Отдельные элементы могут быть получены по одному, просто читая одну строку из потока.
• Входной файл может быть легко разбит на части и разбит по границам строк, передавая каждую часть отдельному рабочему процессу, например, в кластере Hadoop, или просто различным потокам в приложении: вычисление точек разделения, например, путем деления длины на количество рабочих. , затем ищите первую новую строку. Кормите все до этого момента отдельным работником.

Использование StreamReader

Чтобы сделать это, можно использовать TextReader, читать по одной строке за раз и анализировать его с помощью JsonSerializer.Deserialize :

using var reader=new StreamReader(stream);
string line;
//ReadLineAsync() doesn't accept a CancellationToken 
while((line=await reader.ReadLineAsync()) != null)
{
    var item=JsonSerializer.Deserialize<T>(line);
    yield return item;

    if(cancellationToken.IsCancellationRequested)
    {
        return;
    }
}

Это намного проще, чем код, десериализующий правильный массив. Есть две проблемы:

• ReadLineAsync не принимает токен отмены
• Каждая итерация выделяет новую строку, одну из вещей, которую мы хотели избежать , используя System.Text.Json

Этого может быть достаточно, хотя попытка создать ReadOnlySpan<Byte>буферы, необходимые для JsonSerializer.eserialize, не тривиальна.

Трубопроводы и SequenceReader

Чтобы избежать размещения, нам нужно получить ReadOnlySpan<byte>поток. Для этого необходимо использовать каналы System.IO.Pipeline и структуру SequenceReader . В книге Стива Гордона « Введение в SequenceReader» объясняется, как этот класс можно использовать для чтения данных из потока с использованием разделителей.

К сожалению, SequenceReaderэто структура ref, которая означает, что ее нельзя использовать в асинхронных или локальных методах. Вот почему Стив Гордон в своей статье создает

private static SequencePosition ReadItems(in ReadOnlySequence<byte> sequence, bool isCompleted)

Метод чтения элементов формирует ReadOnlySequence и возвращает конечную позицию, поэтому PipeReader может возобновить ее. К сожалению, мы хотим вернуть IEnumerable или IAsyncEnumerable, а методы итератора не любят inили outпараметры либо.

Мы могли бы собрать десериализованные элементы в списке или очереди и вернуть их как единый результат, но это все равно выделило бы списки, буферы или узлы и пришлось бы ждать десериализации всех элементов в буфере перед возвратом:

private static (SequencePosition,List<T>) ReadItems(in ReadOnlySequence<byte> sequence, bool isCompleted)

Нам нужно что-то, что действует как перечислимое, не требуя метода итератора, работает с асинхронностью и не буферизирует все как есть.

Добавление каналов для создания IAsyncEnumerable

ChannelReader.ReadAllAsync возвращает IAsyncEnumerable. Мы можем вернуть ChannelReader из методов, которые не могли работать как итераторы и по-прежнему генерировать поток элементов без кэширования.

Адаптируя код Стива Гордона для использования каналов, мы получаем ReadItems (ChannelWriter ...) и ReadLastItemметоды. Первый, читает по одному элементу за раз, вплоть до новой строки ReadOnlySpan<byte> itemBytes. Это может быть использовано JsonSerializer.Deserialize. Если ReadItemsне удается найти разделитель, он возвращает свою позицию, чтобы PipelineReader мог извлечь следующий фрагмент из потока.

Когда мы достигаем последнего блока и другого разделителя нет, ReadLastItem` читает оставшиеся байты и десериализует их.

Код почти идентичен коду Стива Гордона. Вместо записи в консоль, мы пишем в ChannelWriter.

private const byte NL=(byte)'\n';
private const int MaxStackLength = 128;

private static SequencePosition ReadItems<T>(ChannelWriter<T> writer, in ReadOnlySequence<byte> sequence, 
                          bool isCompleted, CancellationToken token)
{
    var reader = new SequenceReader<byte>(sequence);

    while (!reader.End && !token.IsCancellationRequested) // loop until we've read the entire sequence
    {
        if (reader.TryReadTo(out ReadOnlySpan<byte> itemBytes, NL, advancePastDelimiter: true)) // we have an item to handle
        {
            var item=JsonSerializer.Deserialize<T>(itemBytes);
            writer.TryWrite(item);            
        }
        else if (isCompleted) // read last item which has no final delimiter
        {
            var item = ReadLastItem<T>(sequence.Slice(reader.Position));
            writer.TryWrite(item);
            reader.Advance(sequence.Length); // advance reader to the end
        }
        else // no more items in this sequence
        {
            break;
        }
    }

    return reader.Position;
}

private static T ReadLastItem<T>(in ReadOnlySequence<byte> sequence)
{
    var length = (int)sequence.Length;

    if (length < MaxStackLength) // if the item is small enough we'll stack allocate the buffer
    {
        Span<byte> byteBuffer = stackalloc byte[length];
        sequence.CopyTo(byteBuffer);
        var item=JsonSerializer.Deserialize<T>(byteBuffer);
        return item;        
    }
    else // otherwise we'll rent an array to use as the buffer
    {
        var byteBuffer = ArrayPool<byte>.Shared.Rent(length);

        try
        {
            sequence.CopyTo(byteBuffer);
            var item=JsonSerializer.Deserialize<T>(byteBuffer);
            return item;
        }
        finally
        {
            ArrayPool<byte>.Shared.Return(byteBuffer);
        }

    }    
}

DeserializeToChannel<T>Метод создает читатель трубопроводов на верхней части потока, создает канал и начинает задачу работника , который разбирает ломти и толкают их на канал:

ChannelReader<T> DeserializeToChannel<T>(Stream stream, CancellationToken token)
{
    var pipeReader = PipeReader.Create(stream);    
    var channel=Channel.CreateUnbounded<T>();
    var writer=channel.Writer;
    _ = Task.Run(async ()=>{
        while (!token.IsCancellationRequested)
        {
            var result = await pipeReader.ReadAsync(token); // read from the pipe

            var buffer = result.Buffer;

            var position = ReadItems(writer,buffer, result.IsCompleted,token); // read complete items from the current buffer

            if (result.IsCompleted) 
                break; // exit if we've read everything from the pipe

            pipeReader.AdvanceTo(position, buffer.End); //advance our position in the pipe
        }

        pipeReader.Complete(); 
    },token)
    .ContinueWith(t=>{
        pipeReader.Complete();
        writer.TryComplete(t.Exception);
    });

    return channel.Reader;
}

ChannelReader.ReceiveAllAsync()может быть использован для потребления всех предметов через IAsyncEnumerable<T>:

var reader=DeserializeToChannel<MyEvent>(stream,cts.Token);
await foreach(var item in reader.ReadAllAsync(cts.Token))
{
    //Do something with it 
}    

Такое ощущение, что вам нужно реализовать свой собственный потоковый ридер. Вы должны прочитать байты один за другим и остановиться, как только определение объекта будет завершено. Это действительно довольно низкоуровневый. Таким образом, вы НЕ БУДЕТЕ загружать весь файл в ОЗУ, а просто возьмете на себя роль, с которой имеете дело. Кажется ли это ответом?

Может быть, вы могли бы использовать Newtonsoft.Jsonсериализатор? https://www.newtonsoft.com/json/help/html/Performance.htm

Особенно смотри раздел:

Оптимизировать использование памяти

редактировать

Вы можете попробовать десериализовать значения из JsonTextReader, например

using (var textReader = new StreamReader(stream))
using (var reader = new JsonTextReader(textReader))
{
    while (await reader.ReadAsync(cancellationToken))
    {
        yield return reader.Value;
    }
}