2020年的最后一天，闲着没事上网看了看IAsyncEnumerable。记录一下一些心得。

异步流（IAsyncEnumerable）是作为.net core 3.0/C#8.0的一部分于2019年下半年发布的。当我第一次看到这个名字的时候，我脑海里的第一反应是大神Jeffery Ritcher在08年写的AsyncEnumerable，也就是async/await实现模型的前身。虽然这两个东西名字相近，但它们是在不同年代解决不同的问题。异步流主要解决如何在C#中更方便高效的进行异步流处理（迭代）的问题。

编外话：有兴趣的朋友可以去Channel9搜一下Jeffery Ritcher和AsyncEnumerable的视频，对理解C#的异步模型有很大的帮助。

异步流主要使用了三个新的interface:IAsyncEnumerable, IAsyncEnumerator, IAsyncDisposable。.Net团队在C#中为异步流一如既往的加入了足够的语法糖，这使得异步流使用起来非常方便。从下面的例子可以看出，除了和异步相关的关键词（async和await foreach）之外语法和同步的foreach迭代方式基本一致：

// 定义(下面的例子将TextReader.ReadLineAsync包装成IAsyncEnumerable)
public async IAsyncEnumerable<string?> ReadLineAsync()
{
    string? line = string.Empty;
    while(line != null)
    {
        line = await this.reader.ReadLineAsync();
        if (line != null)
        {
            yield return line;
        }
    }
}

// 用await foreach调用
await foreach(var line in reader.ReadLineAsync()) Use(line)

语法中还可以方便的支持CancellationToken。这些我们不在这里赘述。对于上面的ReadLineAsync，C#会自动生成相应的异步迭代器IAsyncIterator，然后await foreach会在一个while循环中持续调用迭代器的MoveNextAsync直到迭代结束。

我们下面会详细讨论异步流迭代的具体实现细节，效率，以及适合哪些场景下的应用。

C#编译器是如何编译上述语法的

说到这里有两个相关背景要提一下：1. 对于同步的迭代(使用IEnumerable返回值和yield return的函数)，C#编译器会自动根据函数生成一个基于状态机的迭代器用来管理每一步结果的yield return。2. 对于async/await，C#编译器也会自动生成一个状态机管理异步操作完成后的后续代码的回调。相关的知识点网上可以找到很多，我们同样不在这里赘述。

在处理IAsyncEnumerable时，C#编译器将上面两个状态机有机的结合在了一起用来生成相应的IAsyncEnumerator。这个状态机可能的返回情况有三种：有同步当前值可以直接返回时，无当前值开始异步调用需要等待时，迭代结束时（正常或者错误）。

如果拿上面的示例程序做例子，相应的调用流程大致如下：

1. caller开始调用await foreach(...)
2. 自动生成并返回本次的迭代器(IAsyncEnumerator)示例
3. 调用迭代器的GetNextAsync
4. GetNextAsync重置ManualResetValueTaskSource
5. GetNextAsync调用迭代器状态机（通过调用自动生成的IAsyncStateMachine.MoveNext）
1. 如果有值可以同步返回， 保存值到Current，置ManualResetValueTaskSource为true，立即返回。
2. 如果没有当前值可用，状态机调用用户的异步代码。如果代码返回同步值，回到4.a；否则，通过AwaitUnsafeOnComplete设置ManualResetValueTaskSource到状态机的回调然后返回
6. 如果5有同步值返回，GetNextAsync返回值为true的ValueTask；否则返回一个基于ManualResetValueTaskSource的ValueTask。此ValueTaskSource和5中的是同一实例。
7. caller处理GetNextAsync返回值：
1. caller如果拿到ValueTask值为true，访问迭代器的Current拿到当前迭代值，处理并从3重复处理
2. 如果拿到的ValueTask有ValueTaskSource需要等待，caller设置自己的异步状态机现场并向上层返回。
8. 5.b中的异步操作结束，迭代器状态机恢复，拿到当前值，做5.a同样的处理
9. caller在7中的异步状态机恢复，重复7.a的处理。

可以看到，对上面几条简单的几条语句编译器做了非常多的处理。所有这些处理对我们都是透明的，大大简化了程序员的工作。一般情况下我们可以不用担心这些具体细节。

很感谢.Net团队在这方面的辛苦工作。然而，从另外一个角度上，这些编译器的自动代码生成又会导致我们编译后的代码膨胀。这和C#异步编程的实现方式有直接的关系。希望有一天.Net可以直接使用async/await相关的代码信息进行goroutine一样的直接调度而不是使用编译器添加的隐式异步callback。

IAsyncEnumerable相关的效率优化

非常值得一提的是.Net团队在实现IAsyncEnumerable时考虑了非常多的效率优化问题，尤其是对于堆分配优化到了极致。值得一提的是以下两点：

• object复用：

上述操作中用到了很多不同的接口。所有这些接口都被同一个自动生成的class实现，避免了运行时需要分配多个object。甚至在调用GetAsyncEnumerator时，如果没有其它调用，当前的object会被用来直接作为迭代器。当然这是在Solid设计原则和效率之间的有意识妥协。上述的ReadLineAsync对应的类实现如下图所示：

[CompilerGenerated]
private sealed class <ReadLineAsync>d__3 : 
	IAsyncEnumerable<string>,
  IAsyncEnumerator<string>,
  IAsyncDisposable, 
  IValueTaskSource<bool>,
  IValueTaskSource,
  IAsyncStateMachine
{...}

• 使用ValueTask和IValueTaskSource避免无必要的Task对象的分配：

迭代器IAsyncEnumerator的MoveNextAsync原型如下：

ValueTask<bool> MoveNextAsync();

可以看到返回值是一个ValueTask<bool>而不是Task<bool>。如果MoveNextAsync可以有值可以同步返回，我们只需要在栈上分配一个ValueTask<bool>(true)然后返回。如果需要异步等待，一般情况下我们需要对每次异步迭代分配一个Task<bool>然后封装到ValueTask<bool>里面。C#团队没有这么做，他们直接使用了IValueTaskSource，然后通过每次迭代重置它使得代码可以对每次迭代甚至DisposeAsync都重复使用同一个IValueTaskSource。

这些优化加起来的结果就是，不管异步流需要迭代一千次还是一万次，绝大多数情况下自动生成的代码只会在堆上分配一个对象！

异步流的可能应用场景

通过以上可以看出，C#异步流非常简单易用，而且本身在内存方面的效率优化做的很好。这就意味着它很适用于高频次的流式处理，例如：

• 网络协议如TCP/HTTP2数据包/流处理
• 应用层流协议处理，如SignalR，gRPC等
• 数据库io，EFCore据说在考虑用异步流重新部分功能
• 流格式转换，使用异步流可以很轻松高效的将一个流格式转换为新的流格式并且提供非常友好的调用方式

那么WebApi和RESTful怎么样呢？在Webapi里使用异步流可以降低同步处理线程block的风险。RESTful api可以直接返回IAsyncEnumerable。然而，目前的json序列化还需要cache所有数据，所以这个对Webapi性能的影响并不大。

我已经迫不及待想在下一个工作里实验一下异步流了。你呢？

Happy coding, Peace.