在ASP.NET WebAPI 2中流式传输期间，在JSON流中逐个处理JSON对象

Question

我正在尝试流式传输大型JSON文件并在流式传输期间逐项反序列化。

我正在使用这个测试https://raw.githubusercontent.com/ysharplanguage/FastJsonParser/master/JsonTest/TestData/fathers.json.txt。

问题是我没有收到错误，但我的代码似乎没有逐个处理项目，甚至没有处理JSON对象。

我陷入困境，实际上不知道如何实现按对象处理流对象的部分。

这是我的代码：

using Newtonsoft.Json;
using System.Collections.Generic;
using System.IO;
using System.Net;
using System.Net.Http;
using System.Web.Http;
using System.Threading.Tasks;

namespace AMServices.Controllers
{
    public class FathersData
    {
        public Father[] fathers { get; set; }
    }

    public class Someone
    {
        public string name { get; set; }
    }

    public class Father : Someone
    {
        public int id { get; set; }
        public bool married { get; set; }
        // Lists...
        public List<Son> sons { get; set; }
        // ... or arrays for collections, that's fine:
        public Daughter[] daughters { get; set; }
    }

    public class Child : Someone
    {
        public int age { get; set; }
    }

    public class Son : Child
    {
    }

    public class Daughter : Child
    {
        public string maidenName { get; set; }
    }

    public class StreamerController : ApiController
    {
        static readonly JsonSerializer _serializer = new JsonSerializer();
        static readonly HttpClient _client = new HttpClient();

        [HttpPost]
        [Route("streamer/stream")]
        public async Task<IHttpActionResult> stream()
        {
            string apiUrl = "https://raw.githubusercontent.com/ysharplanguage/FastJsonParser/master/JsonTest/TestData/fathers.json.txt";

            using (var stream = await _client.GetStreamAsync(apiUrl).ConfigureAwait(false))
            using (var reader = new StreamReader(stream))
            using (var json = new JsonTextReader(reader))
            {
                if (json == null)
                    StatusCode(HttpStatusCode.InternalServerError);

                JsonSerializer serializer = new JsonSerializer();

                Father f = serializer.Deserialize<Father>(json);
                System.Diagnostics.Debug.WriteLine(f.name);    
            }

            return StatusCode(HttpStatusCode.OK);
        }
    }
}

更新

我已将using (var json = new JsonTextReader(reader))修改为

while (json.Read())
{
    if (json.TokenType == JsonToken.StartObject)
    {
        JObject objX = JObject.Load(json);
        Father f = objX.ToObject<Father>();

        System.Diagnostics.Debug.WriteLine("Name -> " + f.name);
    }
}

我如何逐项更改我可以流和在此过程中？

Answer 1

由于async Enumerable尚不可用，您可以使用回调系统。在下面的示例中，我在一个类中包含反序列化逻辑，它将公开一个事件（qazxsw poi）。

FatherReaded

相同但具有可观察性（Rx）。

async Task Main()
{
    await stream();
}

// Define other methods and classes here
public class FathersData
{
    public Father[] Fathers { get; set; }
}

public class Someone
{
    public string Name { get; set; }
}

public class Father : Someone
{
    public int Id { get; set; }
    public bool Married { get; set; }
    // Lists...
    public List<Son> Sons { get; set; }
    // ... or arrays for collections, that's fine:
    public Daughter[] Daughters { get; set; }
}

public class Child : Someone
{
    public int age { get; set; }
}

public class Son : Child
{
}

public class Daughter : Child
{
    public string maidenName { get; set; }
}

public async Task stream()
{

    var fatherReader = new FatherReader();
    fatherReader.FatherReaded += (s, f) => {
        //f.name.Dump();
        System.Diagnostics.Debug.WriteLine(f.Name);
    };

    string apiUrl = "https://raw.githubusercontent.com/ysharplanguage/FastJsonParser/master/JsonTest/TestData/fathers.json.txt";

    using (var client = new HttpClient())
    using (var stream = await client.GetStreamAsync(apiUrl).ConfigureAwait(false))
    {
        fatherReader.Read(stream);
    }
}

public class FatherReader
{
    public event System.EventHandler<Father> FatherReaded;

    public FatherReader()
    {

    }

    private void OnFatherReaded(Father father){
        FatherReaded?.Invoke(this, father);
    }

    public void Read(Stream stream)
    {
        using (var reader = new StreamReader(stream))
        using (var jsonReader = new JsonTextReader(reader))
        {
            JsonSerializer serializer = new JsonSerializer();

            jsonReader.Read(); // Skip the first StartObject token

            while (jsonReader.Read())
            {
                if (jsonReader.TokenType == JsonToken.StartObject)
                {
                    var father = serializer.Deserialize(jsonReader, typeof(Father));
                    OnFatherReaded((Father)father);
                }
            }
        }
    }
}

方法是Dump的扩展速记

Console.WriteLine(this.ToString)

Answer 2

您可以尝试添加

public async Task stream()
{
    var fatherReader = new FatherReader();
    var observable = fatherReader.Observable;

    // Here you can chain many operator like Linq (filtre, transforme, ...)
    observable = observable
        .Where(f => f.Name.StartsWith("J"));

    observable.Subscribe(f => f.Name.Dump(), e => e.ToString().Dump());

    string apiUrl = "https://raw.githubusercontent.com/ysharplanguage/FastJsonParser/master/JsonTest/TestData/fathers.json.txt";

    using (var client = new HttpClient())
    using (var stream = await client.GetStreamAsync(apiUrl).ConfigureAwait(false))
    {
        fatherReader.Read(stream);
    }
}

public class FatherReader
{
    private Subject<Father> _observable = new Subject<Father>();
    public IObservable<Father> Observable => _observable.AsObservable();

    public FatherReader()
    {
    }

    private void OnFatherReaded(Father father)
    {
        _observable.OnNext(father);
    }

    public void Read(Stream stream)
    {
        try
        {
            using (var reader = new StreamReader(stream))
            using (var jsonReader = new JsonTextReader(reader))
            {
                JsonSerializer serializer = new JsonSerializer();

                jsonReader.Read(); // Skip the first StartObject token

                while (jsonReader.Read())
                {
                    if (jsonReader.TokenType == JsonToken.StartObject)
                    {
                        try
                        {
                            var father = serializer.Deserialize(jsonReader, typeof(Father));
                            OnFatherReaded((Father)father);
                        }
                        catch (Exception ex)
                        {
                            _observable.OnError(ex);
                        }
                    }
                }
            }
        }
        catch (Exception ex)
        {
            _observable.OnError(ex);
        }

        _observable.OnCompleted();
    }
}

并包含RootObject属性作为您的Deserialize对象，因为您的URL JSON数据List<Father>键包含数组而不是对象。

"father"

Api就是这样使用的。

public class RootObject
{
    public List<Father> fathers { get; set; }
}