一篇文章说通C#中的异步迭代器
今天来写写C#中的异步迭代器 - 机制、概念和一些好用的特性
迭代器的概念
迭代器的概念在C#中出现的比较早，很多人可能已经比较熟悉了。
通常迭代器会用在一些特定的场景中。
举个例子：有一个foreach循环：
foreach (var item in Sources)
{
Console.WriteLine(item);
}
这个循环实现了一个简单的功能：把Sources中的每一项在控制台中打印出来。
有时候，Sources可能会是一组完全缓存的数据，例如：List<string>：
IEnumerable<string> Sources(int x)
{
var list = new List<string>();
for (int i = 0; i < 5; i++)
list.Add($"result from Sources, x={x}, result {i}");
return list;
}
这里会有一个小问题：在我们打印Sources的第一个的数据之前，要先运行完整运行Sources()方法来准备数据，在实际应用中，这可能会花费大量时间和内存。更有甚者，Sources可能是一个无边界的列表，或者不定长的开放式列表，比方一次只处理一个数据项目的队列，或者本身没有逻辑结束的队列。
这种情况，C#给出了一个很好的迭代器解决：
IEnumerable<string> Sources(int x)
{
for (int i = 0; i < 5; i++)
yield return $"result from Sources, x={x}, result {i}";
}
这个方式的工作原理与上一段代码很像，但有一些根本的区别 - 我们没有用缓存，而只是每次让一个元素可用。
为了帮助理解，来看看foreach在编译器中的解释：
using (var iter = Sources.GetEnumerator())
{
while (iter.MoveNext())
{
var item = iter.Current;
Console.WriteLine(item);
}
}
当然，这个是省略掉很多东西后的概念解释，我们不纠结这个细节。但大体的意思是这样的：编译器对传递给foreach的表达式调用GetEnumerator()，然后用一个循环去检查是否有下一个数据（MoveNext()），在得到肯定答案后，前进并访问Current属性。而这个属性代表了前进到的元素。
上面这个例子，我们通过MoveNext()/Current方式访问了一个没有大小限制的向前的列表。我们还用到了yield迭代器这个很复杂的东西 - 至少我是这么认为的。
我们把上面的例子中的yield去掉，改写一下看看：
IEnumerable<string> Sources(int x) => new GeneratedEnumerable(x);
class GeneratedEnumerable : IEnumerable<string>
{
private int x;
public GeneratedEnumerable(int x) => this.x = x;
public IEnumerator<string> GetEnumerator() => new GeneratedEnumerator(x);
IEnumerator IEnumerable.GetEnumerator() => GetEnumerator();
}
class GeneratedEnumerator : IEnumerator<string>
{
private int x, i;
public GeneratedEnumerator(int x) => this.x = x;
public string Current { get; private set; }
object IEnumerator.Current => Current;
public void Dispose() { }
public bool MoveNext()
{
if (i < 5)
{
Current = $"result from Sources, x={x}, result {i}";
i++;
return true;
}
else
{
return false;
}
}
void IEnumerator.Reset() => throw new NotSupportedException();
}
这样写完，对照上面的yield迭代器，理解工作过程就比较容易了：
首先，我们给出一个对象IEnumerable。注意，IEnumerable和IEnumerator是不同的。
当我们调用Sources时，就创建了GeneratedEnumerable。它存储状态参数x，并公开了需要的IEnumerable方法。
后面，在需要foreach迭代数据时，会调用GetEnumerator()，而它又调用GeneratedEnumerator以充当数据上的游标。
MoveNext()方法逻辑上实现了for循环，只不过，每次调用MoveNext()只执行一步。更多的数据会通过Current回传过来。另外补充一点：MoveNext()方法中的return false对应于yield break关键字，用于终止迭代。
是不是好理解了？
下面说说异步中的迭代器。
异步中的迭代器
上面的迭代，是同步的过程。而现在Dotnet开发工作更倾向于异步，使用async/await来做，特别是在提高服务器的可伸缩性方面应用特别多。
上面的代码最大的问题，在于MoveNext()。很明显，这是个同步的方法。如果它运行需要一段时间，那线程就会被阻塞。这会让代码执行过程变得不可接受。
我们能做得最接近的方法是异步获取数据：
async Task<List<string>> Sources(int x) {...}
但是，异步获取数据并不能解决数据缓存延迟的问题。
好在，C#为此特意增加了对异步迭代器的支持：
public interface IAsyncEnumerable<out T>
{
IAsyncEnumerator<T> GetAsyncEnumerator(CancellationToken cancellationToken = default);
}
public interface IAsyncEnumerator<out T> : IAsyncDisposable
{
T Current { get; }
ValueTask<bool> MoveNextAsync();
}
public interface IAsyncDisposable
{
ValueTask DisposeAsync();
}
注意，从.NET Standard 2.1和.NET Core 3.0开始，异步迭代器已经包含在框架中了。而在早期版本中，需要手动引入：
# dotnet add package Microsoft.Bcl.AsyncInterfaces
目前这个包的版本号是5.0.0。
还是上面例子的逻辑：
IAsyncEnumerable<string> Source(int x) => throw new NotImplementedException();
看看foreach可以await后的样子：
await foreach (var item in Sources)
{
Console.WriteLine(item);
}
编译器会将它解释为：
await using (var iter = Sources.GetAsyncEnumerator())
{
while (await iter.MoveNextAsync())
{
var item = iter.Current;
Console.WriteLine(item);
}
}
这儿有个新东西：await using。与using用法相同，但释放时会调用DisposeAsync，而不是Dispose，包括回收清理也是异步的。
这段代码其实跟前边的同步版本非常相似，只是增加了await。但是，编译器会分解并重写异步状态机，它就变成异步的了。原理不细说了，不是本文关注的内容。
那么，带有yield的迭代器如何异步呢？看代码：
async IAsyncEnumerable<string> Sources(int x)
{
for (int i = 0; i < 5; i++)
{
await Task.Delay(100); // 这儿模拟异步延迟
yield return $"result from Sources, x={x}, result {i}";
}
}
嗯，看着就舒服。
这就完了？图样图森破。异步有一个很重要的特性：取消。
那么，怎么取消异步迭代？
异步迭代的取消
异步方法通过CancellationToken来支持取消。异步迭代也不例外。看看上面IAsyncEnumerator<T>的定义，取消标志也被传递到了GetAsyncEnumerator()方法中。
那么，如果是手工循环呢？我们可以这样写：
await foreach (var item in Sources.WithCancellation(cancellationToken).ConfigureAwait(false))
{
Console.WriteLine(item);
}
这个写法等同于：
var iter = Sources.GetAsyncEnumerator(cancellationToken);
await using (iter.ConfigureAwait(false))
{
while (await iter.MoveNextAsync().ConfigureAwait(false))
{
var item = iter.Current;
Console.WriteLine(item);
}
}
没错，ConfigureAwait也适用于DisposeAsync()。所以最后就变成了：
await iter.DisposeAsync().ConfigureAwait(false);
异步迭代的取消捕获做完了，接下来怎么用呢？
看代码：
IAsyncEnumerable<string> Sources(int x) => new SourcesEnumerable(x);
class SourcesEnumerable : IAsyncEnumerable<string>
{
private int x;
public SourcesEnumerable(int x) => this.x = x;
public async IAsyncEnumerator<string> GetAsyncEnumerator(CancellationToken cancellationToken = default)
{
for (int i = 0; i < 5; i++)
{
await Task.Delay(100, cancellationToken); // 模拟异步延迟
yield return $"result from Sources, x={x}, result {i}";
}
}
}
如果有CancellationToken通过WithCancellation传过来，迭代器会在正确的时间被取消 - 包括异步获取数据期间（例子中的Task.Delay期间）。当然我们还可以在迭代器中任何一个位置检查IsCancellationRequested或调用ThrowIfCancellationRequested()。
此外，编译器也会通过[EnumeratorCancellation]来完成这个任务，所以我们还可以这样写：
async IAsyncEnumerable<string> Sources(int x, [EnumeratorCancellation] CancellationToken cancellationToken = default) 【文章出处:香港服务器】{ for (int i = 0; i < 5; i++) { await Task.Delay(100, cancellationToken); // 模拟异步延迟 yield return $"result from Sources, x={x}, result {i}"; } }
这个写法与上面的代码其实是一样的，区别在于加了一个参数。
实际应用中，我们有下面几种写法上的选择：
// 不取消 await foreach (var item in Sources) // 通过WithCancellation取消 await foreach (var item in Sources.WithCancellation(cancellationToken)) // 通过SourcesAsync取消 await foreach (var item in SourcesAsync(cancellationToken)) // 通过SourcesAsync和WithCancellation取消 await foreach (var item in SourcesAsync(cancellationToken).WithCancellation(cancellationToken)) // 通过不同的Token取消 await foreach (var item in SourcesAsync(tokenA).WithCancellation(tokenB))
几种方式区别于应用场景，实质上没有区别。对两个Token的方式，任何一个Token被取消时，任务会被取消。
总结
同步迭代其实在各个代码中用的都比较多，但异步迭代用得很好。一方面，这是个相对新的东西，另一方面，是会有点绕，所以很多人都不敢碰。
今天这个，也是个人的一些经验总结，希望对大家理解迭代能有所帮助。
到此这篇关于C#中异步迭代器的文章就介绍到这了,更多相关C#中异步迭代器内容请搜索海外IDC网以前的文章或继续浏览下面的相关文章希望大家以后多多支持海外IDC网！