标签:oid catch red 异步查询 turn ams err 消费 find
C# 8.0中,提供了一种新的IAsyncEnumerable<T>
接口,在对集合进行迭代时,支持异步操作。比如在读取文本中的多行字符串时,如果读取每行字符串的时候使用同步方法,那么会导致线程堵塞。IAsyncEnumerable<T>
可以解决这种情况,在迭代的时候支持使用异步方法。也就是说,之前我们使用foreach
来对IEnumerable
进行迭代,现在可以使用await foreach
来对IAsyncEnumerable<T>
进行迭代,每个项都是可等待的。这种新的接口称为async-streams
,将会随.NET Core 3
发布。我们来看一下如何在LINQ
中实现异步的迭代。
IEnumerable<T>
首先我们创建一个新的Console
项目,基于.NET Core 3
:
namespace AsyncLinqDemo { class Program { static void Main(string[] args) { Console.WriteLine("Input the file path:"); var file = Console.ReadLine(); var lines = ReadAllLines(file); foreach (var line in lines) { Console.WriteLine(line); } } static IEnumerable<string> ReadAllLines(string file) { using (var fs = File.OpenRead(file)) { using (var sr = new StreamReader(fs)) { while (true) { string line = sr.ReadLine(); if(line == null) { break; } yield return line; } } } } } }
这是一个很简单的Console程序,实现了一个简单的返回类型为IEnumerable<string>
的ReadAllLines(string file)
方法,从文本文件中逐行读取文本,并逐行输出。如果文本内容较少的话,没什么问题。但如果我们使用过aync/await
,就会了解,在IO操作如读取或写入文件的时候,最好使用异步方法以避免线程阻塞。让我们来改进一下。
IAsyncEnumerable<T>
可以优化的是下面这句:
string line = sr.ReadLine();
对于IO操作,最好使用异步方式。这里可使用相应的异步方法:
string line = await sr.ReadLineAsync();
我们说“异步是传染的”,如果这里使用异步,那么相应的该方法的返回值也要使用异步,所以需要将返回值改为static async Task<IEnumerable<string>>
,但这样会得到一个错误:
ErrorCS1624The body of ‘Program.ReadAllLines(string)‘ cannot be an iterator block because ‘Task<IEnumerable<string>>‘ is not an iterator interface typeAsyncLinqDemoC:\Source\Workspaces\Console\AsyncLinqDemo\AsyncLinqDemo\Program.cs23Active
因为Task<IEnumerable<string>>
并不是一个可以迭代的接口类型,所以我们无法在方法内部使用yield
关键字。解决问题的办法是使用新的IAsyncEnumerable
接口:
static async IAsyncEnumerable<string> ReadAllLines(string file) { using (var fs = File.OpenRead(file)) { using (var sr = new StreamReader(fs)) { while (true) { string line = await sr.ReadLineAsync(); if(line == null) { break; } yield return line; } } } }
按F12
查看该接口的定义:
namespace System.Collections.Generic { public interface IAsyncEnumerable<out T> { IAsyncEnumerator<T> GetAsyncEnumerator(CancellationTokencancellationToken = default); } }
这是一个异步的迭代器,并提供了CancellationToken
。再按F12
查看IAsyncEnumerator<T>
的定义,可发现里面是这样的:
namespace System.Collections.Generic { public interface IAsyncEnumerator<out T> : IAsyncDisposable { T Current { get; } ValueTask<bool> MoveNextAsync(); } }
这里MoveNextAsync()
方法实际是返回了一个结果类型为bool
的Task
,每次迭代都是可等待的,从而实现了迭代器的异步。
await foreach
消费IAsyncEnumerable<T>
当我们做了以上改动之后,ReadAllLines()
方法返回的是一个支持异步的IAsyncEnumerable
,那么在使用的时候,也不能简单的使用foreach
了。修改Main
方法如下:
static async Task Main(string[] args) { Console.WriteLine("Input the file path:"); var file = Console.ReadLine(); var lines = ReadAllLines(file); await foreach (var line in lines) { Console.WriteLine(line); } }
首先在foreach
之前添加await
关键字,还要需要将Main
方法由void
改为async Task
。这样整个程序都是异步执行了,不会再导致堵塞了。这个例子只是一个简单的demo,是否使用异步并不会感觉到明显的区别。如果在迭代内部需要比较重的操作,如从网络获取大量数据或读取大量磁盘文件,异步的优势还是会比较明显的。
LINQ
消费IAsyncEnumerable<T>
使用LINQ
来操作集合是常用的功能。如果使用IEnumberable
,在Main
方法中可以做如下改动:
var lines = ReadAllLines(file); var res = from line in lines where line.StartsWith("ERROR: ") selectline.Substring("ERROR: ".Length); foreach (var line in res) { Console.WriteLine(line); }
或:
var res = lines.Where(x => x.StartsWith("ERROR: ")).Select(x => x.Substring("ERROR: ".Length));
如果使用了新的IAsyncEnumerable
,你会发现无法使用Where
等操作符了:
ErrorCS1936Could not find an implementation of the query pattern for source type ‘IAsyncEnumerable<string>‘. ‘Where‘ not found.AsyncLinqDemoC:\Source\Workspaces\Console\AsyncLinqDemo\AsyncLinqDemo\Program.cs16Active
目前LINQ
还没有提供对IAsyncEnumerable
的原生支持,不过微软提供了一个Nuget包来实现此功能。在项目中打开Nuget Package Manger搜索安装System.Linq.Async
,注意该包目前还是预览版,所以要勾选include prerelease
才能看到。安装该Nuget包后,Linq查询语句中的错误就消失了。
在System.Linq.Async
这个包中,对每个同步的LINQ
方法都做了相应的扩展。所以基本上代码无需什么改动即可正常编译。
对于LINQ
中的条件语句,也可以使用WhereAwait()
方法来支持await
:
public static IAsyncEnumerable<TSource> WhereAwait<TSource>(thisIAsyncEnumerable<TSource> source, Func<TSource, int, ValueTask<bool>>predicate);
如需要在条件语句中进行IO或网络请求等异步操作,可以这样用:
var res = lines.WhereAwait(async x => await DoSomeHeavyOperationsAsync(x));
DoSomeHeavyOperationsAsync
方法的签名如下:
private static ValueTask<bool> DoSomeHeavyOperationsAsync(string x) { //Do some works... }
通过以上的示例,我们简要了解了如何使用IAsyncEnumerable
接口以及如何在LINQ
中实现异步查询。在使用该接口时,我们需要创建一个自定义方法返回IAsyncEnumerable<T>
来代替IEnumberable<T>
,这个方法可称为async-iterator
方法,需要注意以下几点:
该方法应该被声明为async
。
返回IAsyncEnumerable<T>
。
同时使用await
及yield
。如await foreach
,yield return
或yield break
等。
例如:
async IAsyncEnumerable<int> GetValuesFromServer() { while (true) { IEnumerable<int> batch = await GetNextBatch(); if (batch == null) yield break; foreach (int item in batch) { yield return item; } } }
此外还有一些限制:
无法在try
的finally
块中使用任何形式的yield
语句。
无法在包含任何catch
语句的try
语句中使用yield return
语句。
期待.NET Core 3的正式发布!
了解新西兰IT行业真实码农生活
请长按上方二维码关注“程序员在新西兰”
C#8.0: 在 LINQ 中支持异步的 IAsyncEnumerable
标签:oid catch red 异步查询 turn ams err 消费 find
原文地址:https://www.cnblogs.com/yanxiaodi/p/Support-IAsyncEnumerable-with-LINQ.html