Desserializando uma lista de forma assíncrona usando System.Text.Json

Digamos que eu solicite um arquivo json grande que contenha uma lista de muitos objetos. Não quero que eles estejam na memória de uma só vez, mas prefiro lê-los e processá-los um por um. Então, eu preciso transformar um System.IO.Streamfluxo assíncrono em um IAsyncEnumerable<T>. Como uso a nova System.Text.JsonAPI para fazer isso?

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
        }
    }
}

Sim, um serializador JSON (de) verdadeiramente de streaming seria uma boa melhoria de desempenho, em muitos lugares.

Infelizmente, System.Text.Jsonnão faz isso no momento. Não tenho certeza se será no futuro - espero que sim! A desserialização do JSON por streaming é realmente um desafio.

Você pode verificar se o Utf8Json extremamente rápido suporta, talvez.

No entanto, pode haver uma solução personalizada para sua situação específica, pois seus requisitos parecem restringir a dificuldade.

A idéia é ler manualmente um item da matriz por vez. Estamos fazendo uso do fato de que cada item da lista é, por si só, um objeto JSON válido.

Você pode pular manualmente o [(para o primeiro item) ou o ,(para cada próximo item). Acho que sua melhor aposta é usar o .NET Core Utf8JsonReaderpara determinar onde o objeto atual termina e alimentar os bytes digitalizados JsonDeserializer.

Dessa forma, você está armazenando um pouco de buffer de um objeto por vez.

E já que estamos falando de desempenho, você pode obter a entrada de a PipeReaderenquanto está nisso. :-)

TL; DR Não é trivial

Parece que alguém já postou o código completo de uma Utf8JsonStreamReaderestrutura que lê buffers de um fluxo e os alimenta para um Utf8JsonRreader, permitindo fácil desserialização com JsonSerializer.Deserialize<T>(ref newJsonReader, options);. O código também não é trivial. A pergunta relacionada está aqui e a resposta está aqui .

Porém, isso não é suficiente - HttpClient.GetAsyncretornará somente depois que toda a resposta for recebida, armazenando essencialmente tudo na memória.

Para evitar isso, HttpClient.GetAsync (string, HttpCompletionOption) deve ser usado com HttpCompletionOption.ResponseHeadersRead.

O loop de desserialização também deve verificar o token de cancelamento e sair ou lançar se estiver sinalizado. Caso contrário, o loop continuará até que todo o fluxo seja recebido e processado.

Esse código é baseado no exemplo da resposta relacionada e usa HttpCompletionOption.ResponseHeadersReade verifica o token de cancelamento. Ele pode analisar cadeias JSON que contêm uma matriz adequada de itens, por exemplo:

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

A primeira chamada para jsonStreamReader.Read()se move para o início da matriz, enquanto a segunda se move para o início do primeiro objeto. O próprio loop termina quando o final da matriz ( ]) é detectado.

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();
    }
}

Fragmentos JSON, AKA streaming de JSON aka ... *

É bastante comum nos cenários de streaming ou log de eventos anexar objetos JSON individuais a um arquivo, um elemento por linha, por exemplo:

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

Este não é um documento JSON válido , mas os fragmentos individuais são válidos. Isso tem várias vantagens para cenários de big data / altamente simultâneos. A adição de um novo evento requer apenas o acréscimo de uma nova linha ao arquivo, não a análise e a reconstrução do arquivo inteiro. O processamento , especialmente o processamento paralelo , é mais fácil por dois motivos:

• Elementos individuais podem ser recuperados um de cada vez, simplesmente lendo uma linha de um fluxo.
• O arquivo de entrada pode ser facilmente particionado e dividido entre os limites da linha, alimentando cada parte com um processo de trabalho separado, por exemplo, em um cluster Hadoop, ou simplesmente com threads diferentes em um aplicativo: Calcule os pontos de divisão, por exemplo, dividindo o comprimento pelo número de trabalhadores , procure a primeira nova linha. Alimente tudo até esse ponto para um trabalhador separado.

Usando um StreamReader

A maneira de alocar-y para fazer isso seria usar um TextReader, ler uma linha de cada vez e analisá-la com 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;
    }
}

Isso é muito mais simples que o código que desserializa uma matriz adequada. Existem dois problemas:

• ReadLineAsync não aceita um token de cancelamento
• Cada iteração aloca uma nova string, uma das coisas que queríamos evitar usando System.Text.Json

Isso pode ser suficiente, pois a tentativa de produzir os ReadOnlySpan<Byte>buffers necessários ao JsonSerializer.Deserialize não é trivial.

Pipelines e SequenceReader

Para evitar alocações, precisamos obter a ReadOnlySpan<byte>partir do fluxo. Isso requer o uso de pipes System.IO.Pipeline e a estrutura SequenceReader . Uma Introdução ao SequenceReader, de Steve Gordon, explica como essa classe pode ser usada para ler dados de um fluxo usando delimitadores.

Infelizmente, SequenceReaderé uma ref struct, o que significa que não pode ser usado em métodos assíncronos ou locais. É por isso que Steve Gordon em seu artigo cria um

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

O método para ler itens forma um ReadOnlySequence e retorna a posição final, para que o PipeReader possa retomar a partir dele. Infelizmente , queremos retornar um IEnumerable ou IAsyncEnumerable, e os métodos do iterador também não gostam innem dos outparâmetros.

Poderíamos coletar os itens desserializados em uma Lista ou Fila e devolvê-los como um único resultado, mas isso ainda alocaria listas, buffers ou nós e teríamos que esperar que todos os itens em um buffer fossem desserializados antes de retornar:

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

Precisamos de algo que atue como um enumerável sem a necessidade de um método iterador, que funcione com assíncrono e não armazene em buffer tudo.

Adicionando canais para produzir um IAsyncEnumerable

ChannelReader.ReadAllAsync retorna um IAsyncEnumerable. Podemos retornar um ChannelReader de métodos que não poderiam funcionar como iteradores e ainda produzir um fluxo de elementos sem armazenar em cache.

Adaptando o código de Steve Gordon para usar canais, obtemos os ReadItems (ChannelWriter ...) e ReadLastItemmétodos. O primeiro, lê um item de cada vez, usando uma nova linha ReadOnlySpan<byte> itemBytes. Isso pode ser usado por JsonSerializer.Deserialize. Se ReadItemsnão conseguir encontrar o delimitador, ele retornará sua posição para que o PipelineReader possa extrair o próximo pedaço do fluxo.

Quando atingimos o último pedaço e não há outro delimitador, o ReadLastItem` lê os bytes restantes e os desserializa.

O código é quase idêntico ao de Steve Gordon. Em vez de escrever no console, escrevemos no 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);
        }

    }    
}

O DeserializeToChannel<T>método cria um leitor de pipeline na parte superior do fluxo, cria um canal e inicia uma tarefa de trabalho que analisa os pedaços e os envia ao canal:

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()pode ser usado para consumir todos os itens através de um IAsyncEnumerable<T>:

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

Parece que você precisa implantar seu próprio leitor de fluxo. Você precisa ler os bytes um por um e parar assim que a definição do objeto for concluída. Na verdade, é bastante de baixo nível. Como tal, você NÃO carregará o arquivo inteiro na RAM, mas participará da parte com a qual está lidando. Parece ser uma resposta?

Talvez você possa usar o Newtonsoft.Jsonserializador? https://www.newtonsoft.com/json/help/html/Performance.htm

Veja especialmente a seção:

Otimizar o uso da memória

Editar

Você pode tentar desserializar valores do JsonTextReader, por exemplo

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