- html - 出于某种原因,IE8 对我的 Sass 文件中继承的 html5 CSS 不友好?
- JMeter 在响应断言中使用 span 标签的问题
- html - 在 :hover and :active? 上具有不同效果的 CSS 动画
- html - 相对于居中的 html 内容固定的 CSS 重复背景?
假设我请求一个包含许多对象列表的大型 json 文件。我不希望它们一下子都在内存中,但我宁愿一个一个地阅读和处理它们。所以我需要转一个异步 System.IO.Stream
流入 IAsyncEnumerable<T>
.如何使用新版 System.Text.Json
API来做到这一点?
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
}
}
}
最佳答案
TL;DR 这不是微不足道的
看起来有人已经 posted full code 了 Utf8JsonStreamReader
结构,它从流中读取缓冲区并将它们提供给 Utf8JsonRreader,允许使用 JsonSerializer.Deserialize<T>(ref newJsonReader, options);
轻松反序列化。代码也不是微不足道的。相关问题是 here ,答案是 here 。
但这还不够 - HttpClient.GetAsync
只有在收到整个响应后才会返回,本质上是在内存中缓冲所有内容。
为了避免这种情况,HttpClient.GetAsync(string,HttpCompletionOption ) 应该与 HttpCompletionOption.ResponseHeadersRead
一起使用。
反序列化循环也应该检查取消 token ,如果有信号则退出或抛出。否则循环将继续,直到整个流被接收和处理。
此代码基于相关答案的示例,并使用 HttpCompletionOption.ResponseHeadersRead
并检查取消标记。它可以解析包含适当项目数组的 JSON 字符串,例如:
[{"prop1":123},{"prop1":234}]
jsonStreamReader.Read()
移动到数组的开头,而第二次调用移动到第一个对象的开头。当检测到数组的结尾 (
]
) 时,循环本身终止。
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();
}
}
{"eventId":1}
{"eventId":2}
...
{"eventId":1234567}
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
不接受取消 token ReadOnlySpan<Byte>
缓冲区并非易事。
ReadOnlySpan<byte>
。这样做需要使用 System.IO.Pipeline 管道和
SequenceReader 结构。 Steve Gordon 的
An Introduction to SequenceReader 解释了如何使用这个类使用分隔符从流中读取数据。
SequenceReader
是一个 ref 结构,这意味着它不能用于异步或本地方法。这就是为什么史蒂夫戈登在他的文章中创建了一个
private static SequencePosition ReadItems(in ReadOnlySequence<byte> sequence, bool isCompleted)
in
或
out
参数。
private static (SequencePosition,List<T>) ReadItems(in ReadOnlySequence<byte> sequence, bool isCompleted)
ReadLastItem
方法。第一个,一次读取一个项目,直到使用
ReadOnlySpan<byte> itemBytes
换行。这可以由
JsonSerializer.Deserialize
使用。如果
ReadItems
找不到分隔符,它将返回其位置,以便 PipelineReader 可以从流中提取下一个块。
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>
方法在流的顶部创建一个管道读取器,创建一个 channel 并启动一个工作任务来解析块并将它们推送到 channel :
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
}
关于c# - 使用 System.Text.Json 异步反序列化列表,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/58572524/
我有一个带有帮助页面的 Web API 2 项目,该项目在本地运行良好,但当我将其推送到 Azure 时抛出此错误: Method not found: 'System.String System.S
我有两台服务器,但通常运行相同的设置 - IIS、SQL Server 等。一台给我这个错误,另一台没有。我从 Visual Studio 向两者发布相同的代码。 它们都在运行 .NET CLR Ve
System.out声明为 public static final PrintStream out。 但是你可以调用System.setOut()重新分配它。 嗯?如果它是 final,这怎么可能?
System.out被声明为 public static final PrintStream out。 但是您可以调用System.setOut()重新分配它。 嗯?如果是 final,这怎么可能?
我有这个 linq 查询: private void GetReceivedInvoiceTasks(User user, List tasks) { var areaIds = user.A
我有一个 MonoTouch 应用程序,当我为设备编译它时,出现以下错误: Error MT2002: Can not resolve reference: System.Boolean System
您好,我有一个名为 DailyVisitReport 的 View 。在该 View 中,我有两个名为 FromDate 和 toDate 的字段。如果我选择 FromDate 和 ToDate 取决
是否可以从 ObjectContext 对象中读取元组列表? 我在存储过程中有类似这样的数据库查询 SELECT T.Id as Item1, -- this is guid T.Wo
我正在尝试创建 Odata 端点,但每当我尝试执行任何涉及日期的查询时都会收到此错误。 我在下面的非常简单示例中重新创建了它。 数据库表 EDMX(片段)
我正在尝试创建 Odata 端点,但每当我尝试执行任何涉及日期的查询时都会收到此错误。 我在下面的非常简单示例中重新创建了它。 数据库表 EDMX(片段)
我有一个方法可以从数据读取器的数据中生成类类型列表。 if (datareader != null && datareader .HasRows) { Dictionary pDict= GetP
我有一些旧的 C++ 代码,它们使用 stdio 进行输入和输出。该代码还通过 fork 生成新进程。它将 stdio 重新映射到每个新进程,以便每个 session 获取其各自的数据。 我正在考虑使
我的应用程序可以很好地构建/链接/部署到模拟器,但我只是第一次尝试将应用程序构建/部署到真实设备,并且链接器失败。 我不使用 System.Console或 ConsoleColor在我的应用程序的任
主要是我很好奇。 我们有一个名为 Unit 的对象在我们的代码库中 - 代表桥梁或道路的组件。在我们的例子中,看到带有 Unit 的 ReactiveUI 命令可能会模棱两可。作为声明中的泛型之一。
我试图将Object变量转换为StreamWriter。但是,它不起作用。有什么错? StreamWriter file = (StreamWriter) myObject; 最佳答案 myObjec
为什么以下不编译? using System; using System.Linq; using System.Linq.Expressions; public static class Extens
我正在使用 Visual Studio Community 2015 开发面向 .NET 4.5 的 Visual Basic 应用程序.我没有编写应用程序,所以我使用 NuGet 添加了所有缺失的依
我刚刚开始使用 powershell,我正在制作一个非常简单的加密功能。我想获取字符串中的每个字符,将其转换为 int 并添加一个选定的数字,然后将其转换回一个字符。 这工作正常: function
一些使用我的应用程序的人似乎变得越来越 System.MissingMethodException: Method not found: 'System.Object System.Windows.T
我是 C# 和实体的新手 我想知道是否有人在这里帮助我。我选择了哪个返回我的 customerid,所以我想将它作为参数传递给我的构造函数,我的构造函数参数类型是 guid 但我的选择类型不同,我不知
我是一名优秀的程序员,十分优秀!