行分隔的 json 序列化和反序列化

Question

我正在使用 JSON.NET 和 C# 5。我需要将对象列表序列化/反序列化为行分隔的 json。 http://en.wikipedia.org/wiki/Line_Delimited_JSON。例如，

{"some":"thing1"}
{"some":"thing2"}
{"some":"thing3"}

和

{"kind": "person", "fullName": "John Doe", "age": 22, "gender": "Male", "citiesLived": [{ "place": "Seattle", "numberOfYears": 5}, {"place": "Stockholm", "numberOfYears": 6}]}
{"kind": "person", "fullName": "Jane Austen", "age": 24, "gender": "Female", "citiesLived": [{"place": "Los Angeles", "numberOfYears": 2}, {"place": "Tokyo", "numberOfYears": 2}]}

为什么我需要它，因为它的 Google BigQuery 要求 https://cloud.google.com/bigquery/preparing-data-for-bigquery

更新：我发现的一种方法是单独序列化每个对象并在最后加入换行符。

Answer 1

您可以通过使用 JsonTextReader 手动解析 JSON 并将 SupportMultipleContent 标志设置为 true 来实现。

如果我们查看您的第一个示例，并创建一个名为 Foo 的 POCO：

public class Foo
{
    [JsonProperty("some")]
    public string Some { get; set; }
}

这是我们解析它的方式：

var json = "{\"some\":\"thing1\"}\r\n{\"some\":\"thing2\"}\r\n{\"some\":\"thing3\"}";
var jsonReader = new JsonTextReader(new StringReader(json))
{
    SupportMultipleContent = true // This is important!
};

var jsonSerializer = new JsonSerializer();
while (jsonReader.Read())
{
    Foo foo = jsonSerializer.Deserialize<Foo>(jsonReader);
}

如果您想要项目列表作为结果，只需将每个项目添加到 while 循环内的列表中即可。

listOfFoo.Add(jsonSerializer.Deserialize<Foo>(jsonReader));

注意：对于 Json.Net 10.0.4 及更高版本，相同的代码还支持逗号分隔的 JSON 条目，请参阅How to deserialize dodgy JSON (with improperly quoted strings, and missing brackets)?)

Answer 2

为了使用 .NET 5 (C# 9) 和 System.Text.Json.JsonSerializer 类以及“大”数据来实现，我编写了流处理代码。

使用System.IO.Pipelines扩展包，效率很高。

using System;
using System.Buffers;
using System.Collections.Generic;
using System.IO;
using System.IO.Pipelines;
using System.Text;
using System.Text.Json;
using System.Threading.Tasks;

class Program
{
    static readonly byte[] NewLineChars = {(byte)'\r', (byte)'\n'};
    static readonly byte[] WhiteSpaceChars = {(byte)'\r', (byte)'\n', (byte)' ', (byte)'\t'};

    private static async Task Main()
    {
        JsonSerializerOptions jsonOptions = new(JsonSerializerDefaults.Web);
        var json = "{\"some\":\"thing1\"}\r\n{\"some\":\"thing2\"}\r\n{\"some\":\"thing3\"}";
        var contentStream = new MemoryStream(Encoding.UTF8.GetBytes(json));
        var pipeReader = PipeReader.Create(contentStream);
        await foreach (var foo in ReadItemsAsync<Foo>(pipeReader, jsonOptions))
        {
            Console.WriteLine($"foo: {foo.Some}");
        }
    }

    static async IAsyncEnumerable<TValue> ReadItemsAsync<TValue>(PipeReader pipeReader, JsonSerializerOptions jsonOptions = null)
    {
        while (true)
        {
            var result = await pipeReader.ReadAsync();
            var buffer = result.Buffer;
            bool isCompleted = result.IsCompleted;
            SequencePosition bufferPosition = buffer.Start;
            while (true)
            {
                var(value, advanceSequence) = TryReadNextItem<TValue>(buffer, ref bufferPosition, isCompleted, jsonOptions);
                if (value != null)
                {
                    yield return value;
                }

                if (advanceSequence)
                {
                    pipeReader.AdvanceTo(bufferPosition, buffer.End); //advance our position in the pipe
                    break;
                }
            }

            if (isCompleted)
                yield break;
        }
    }

    static (TValue, bool) TryReadNextItem<TValue>(ReadOnlySequence<byte> sequence, ref SequencePosition sequencePosition, bool isCompleted, JsonSerializerOptions jsonOptions)
    {
        var reader = new SequenceReader<byte>(sequence.Slice(sequencePosition));
        while (!reader.End) // loop until we've come to the end or read an item
        {
            if (reader.TryReadToAny(out ReadOnlySpan<byte> itemBytes, NewLineChars, advancePastDelimiter: true))
            {
                sequencePosition = reader.Position;
                if (itemBytes.TrimStart(WhiteSpaceChars).IsEmpty)
                {
                    continue;
                }

                return (JsonSerializer.Deserialize<TValue>(itemBytes, jsonOptions), false);
            }
            else if (isCompleted)
            {
                // read last item
                var remainingReader = sequence.Slice(reader.Position);
                ReadOnlySpan<byte> remainingSpan = remainingReader.IsSingleSegment ? remainingReader.First.Span : remainingReader.ToArray();
                reader.Advance(remainingReader.Length); // advance reader to the end
                sequencePosition = reader.Position;
                if (!remainingSpan.TrimStart(WhiteSpaceChars).IsEmpty)
                {
                    return (JsonSerializer.Deserialize<TValue>(remainingSpan, jsonOptions), true);
                }
                else
                {
                    return (default, true);
                }
            }
            else
            {
                // no more items in sequence
                break;
            }
        }

        // PipeReader needs to read more
        return (default, true);
    }
}

public class Foo
{
    public string Some
    {
        get;
        set;
    }
}

在https://dotnetfiddle.net/M5cNo1运行