C#8.0: 在 LINQ 中支持异步的 IAsyncEnumerable

雲小雲 發表於 2019-7-12 06:44:00

C#8.0: 在 LINQ 中支持异步的 IAsyncEnumerable

C# 8.0中，提供了一种新的<code>IAsyncEnumerable<T></code>接口，在对集合进行迭代时，支持异步操作。比如在读取文本中的多行字符串时，如果读取每行字符串的时候使用同步方法，那么会导致线程堵塞。<code>IAsyncEnumerable<T></code>可以解决这种情况，在迭代的时候支持使用异步方法。也就是说，之前我们使用<code>foreach</code>来对<code>IEnumerable</code>进行迭代，现在可以使用<code>await foreach</code>来对<code>IAsyncEnumerable<T></code>进行迭代，每个项都是可等待的。这种新的接口称为<code>async-streams</code>，将会随<code>.NET Core 3</code>发布。我们来看一下如何在<code>LINQ</code>中实现异步的迭代。
<h2>使用常规的<code>IEnumerable<T></code></h2>
首先我们创建一个新的<code>Console</code>项目，基于<code>.NET Core 3</code>：
<div class="cnblogs_code">
<pre>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;
 }
 }
 }
 }
}
}</pre>
</div>
 
这是一个很简单的Console程序，实现了一个简单的返回类型为<code>IEnumerable<string></code>的<code>ReadAllLines(string file)</code>方法，从文本文件中逐行读取文本，并逐行输出。如果文本内容较少的话，没什么问题。但如果我们使用过<code>aync/await</code>，就会了解，在IO操作如读取或写入文件的时候，最好使用异步方法以避免线程阻塞。让我们来改进一下。
<h2>使用异步的<code>IAsyncEnumerable<T></code></h2>
可以优化的是下面这句：
<div class="cnblogs_code">
<pre>string line = sr.ReadLine();</pre>
</div>
 
对于IO操作，最好使用异步方式。这里可使用相应的异步方法：
<div class="cnblogs_code">
<pre>string line = await sr.ReadLineAsync();</pre>
</div>
 
我们说“异步是传染的”，如果这里使用异步，那么相应的该方法的返回值也要使用异步，所以需要将返回值改为<code>static async Task<IEnumerable<string>></code>，但这样会得到一个错误：
<div class="cnblogs_code">
<pre>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</pre>
</div>
 
因为<code>Task<IEnumerable<string>></code>并不是一个可以迭代的接口类型，所以我们无法在方法内部使用<code>yield</code>关键字。解决问题的办法是使用新的<code>IAsyncEnumerable</code>接口：
<div class="cnblogs_code">
<pre>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;
 }

 }
}
}</pre>
</div>
 
按<code>F12</code>查看该接口的定义：
<div class="cnblogs_code">
<pre>namespace System.Collections.Generic
{
public interface IAsyncEnumerable<out T>
{
 IAsyncEnumerator<T> GetAsyncEnumerator(CancellationTokencancellationToken = default);
}
}</pre>
</div>
 
这是一个异步的迭代器，并提供了<code>CancellationToken</code>。再按<code>F12</code>查看<code>IAsyncEnumerator<T></code>的定义，可发现里面是这样的：
<div class="cnblogs_code">
<pre>namespace System.Collections.Generic
{
public interface IAsyncEnumerator<out T> : IAsyncDisposable
{
 T Current { get; }
 ValueTask<bool> MoveNextAsync();
}
}</pre>
</div>
 
这里<code>MoveNextAsync()</code>方法实际是返回了一个结果类型为<code>bool</code>的<code>Task</code>，每次迭代都是可等待的，从而实现了迭代器的异步。
<h2>使用<code>await foreach</code>消费<code>IAsyncEnumerable<T></code></h2>
当我们做了以上改动之后，<code>ReadAllLines()</code>方法返回的是一个支持异步的<code>IAsyncEnumerable</code>，那么在使用的时候，也不能简单的使用<code>foreach</code>了。修改<code>Main</code>方法如下：
<div class="cnblogs_code">
<pre>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);
}
}</pre>
</div>
 
首先在<code>foreach</code>之前添加<code>await</code>关键字，还要需要将<code>Main</code>方法由<code>void</code>改为<code>async Task</code>。这样整个程序都是异步执行了，不会再导致堵塞了。这个例子只是一个简单的demo，是否使用异步并不会感觉到明显的区别。如果在迭代内部需要比较重的操作，如从网络获取大量数据或读取大量磁盘文件，异步的优势还是会比较明显的。
<h2>使用<code>LINQ</code>消费<code>IAsyncEnumerable<T></code></h2>
使用<code>LINQ</code>来操作集合是常用的功能。如果使用<code>IEnumberable</code>，在<code>Main</code>方法中可以做如下改动：
<div class="cnblogs_code">
<pre>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);
}</pre>
</div>
 
或：
<div class="cnblogs_code">
<pre>var res = lines.Where(x => x.StartsWith("ERROR: ")).Select(x => x.Substring("ERROR: ".Length));</pre>
</div>
 
如果使用了新的<code>IAsyncEnumerable</code>，你会发现无法使用<code>Where</code>等操作符了：
<div class="cnblogs_code">
<pre>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</pre>
</div>
 
目前<code>LINQ</code>还没有提供对<code>IAsyncEnumerable</code>的原生支持，不过微软提供了一个Nuget包来实现此功能。在项目中打开Nuget Package Manger搜索安装<code>System.Linq.Async</code>，注意该包目前还是预览版，所以要勾选<code>include prerelease</code>才能看到。安装该Nuget包后，Linq查询语句中的错误就消失了。
在<code>System.Linq.Async</code>这个包中，对每个同步的<code>LINQ</code>方法都做了相应的扩展。所以基本上代码无需什么改动即可正常编译。
对于<code>LINQ</code>中的条件语句，也可以使用<code>WhereAwait()</code>方法来支持<code>await</code>：
<div class="cnblogs_code">
<pre>public static IAsyncEnumerable<TSource> WhereAwait<TSource>(thisIAsyncEnumerable<TSource> source, Func<TSource, int, ValueTask<bool>>predicate);</pre>
</div>
 
如需要在条件语句中进行IO或网络请求等异步操作，可以这样用：
<div class="cnblogs_code">
<pre>var res = lines.WhereAwait(async x => await DoSomeHeavyOperationsAsync(x));</pre>
</div>
 
<code>DoSomeHeavyOperationsAsync</code>方法的签名如下：
<div class="cnblogs_code">
<pre>private static ValueTask<bool> DoSomeHeavyOperationsAsync(string x)
{
//Do some works...
}</pre>
</div>
 
<h2>小结</h2>
通过以上的示例，我们简要了解了如何使用<code>IAsyncEnumerable</code>接口以及如何在<code>LINQ</code>中实现异步查询。在使用该接口时，我们需要创建一个自定义方法返回<code>IAsyncEnumerable<T></code>来代替<code>IEnumberable<T></code>，这个方法可称为<code>async-iterator</code>方法，需要注意以下几点：
<ul class="list-paddingleft-2" data-mark="*">
<li>
该方法应该被声明为<code>async</code>。
</li>
<li>
返回<code>IAsyncEnumerable<T></code>。
</li>
<li>
同时使用<code>await</code>及<code>yield</code>。如<code>await foreach</code>，<code>yield return</code>或<code>yield break</code>等。
</li>
</ul>
例如：
<div class="cnblogs_code">
<pre>async IAsyncEnumerable<int> GetValuesFromServer()
{
while (true)
{
 IEnumerable<int> batch = await GetNextBatch();
 if (batch == null) yield break;

 foreach (int item in batch)
 {
 yield return item;
 }
}
}</pre>
</div>
 
此外还有一些限制：
<ul class="list-paddingleft-2" data-mark="*">
<li>
无法在<code>try</code>的<code>finally</code>块中使用任何形式的<code>yield</code>语句。
</li>
<li>
无法在包含任何<code>catch</code>语句的<code>try</code>语句中使用<code>yield return</code>语句。
 
</li>
</ul>
期待.NET Core 3的正式发布！
 
<img style="display: block; margin-left: auto; margin-right: auto" src="https://mmbiz.qpic.cn/mmbiz_jpg/ahAATVIdckRjwCicQictHicK9Ebav2YbXIiblyZ45Q7n50Jv20xzXsj94fueJ8SQcL1ZQibEZLKs1p55kzQiaBFyMDBQ/640?wx_fmt=jpeg&tp=webp&wxfrom=5&wx_lazy=1&wx_co=1" alt="" width="287" height="196" data-copyright="0" data-ratio="0.6822074215033301" data-s="300,640" data-type="jpeg" data-w="1051" data-src="https://mmbiz.qpic.cn/mmbiz_jpg/ahAATVIdckRjwCicQictHicK9Ebav2YbXIiblyZ45Q7n50Jv20xzXsj94fueJ8SQcL1ZQibEZLKs1p55kzQiaBFyMDBQ/640?wx_fmt=jpeg" data-fail="0">
了解新西兰IT行业真实码农生活
请长按上方二维码关注“程序员在新西兰” 
来源：https://www.cnblogs.com/yanxiaodi/p/Support-IAsyncEnumerable-with-LINQ.html

頁: [1]

琼殿技术论坛's Archiver

C#8.0: 在 LINQ 中支持异步的 IAsyncEnumerable