C# 8中的Async Streams

关键要点

• 异步编程技术提供了一种提升程序响应能力的方法。
• Async/Await模式在C# 5中首次亮相，但只能返回单个标量值。
• C# 8添加了异步流（Async Streams），容许异步方法返回多个值，从而扩展了其可用性。
• 异步流提供了一种用于表示异步数据源的绝佳方法。
• 异步流是Java和JavaScript中使用的反应式编程模型的替代方案。

C# 5引入了Async/Await，用以提升用户界面响应能力和对Web资源的访问能力。换句话说，异步方法用于执行不阻塞线程并返回一个标量结果的异步操做。

微软屡次尝试简化异步操做，由于Async/Await模式易于理解，因此在开发人员当中得到了良好的承认。

现有异步方法的一个重要不足是它必须提供一个标量返回结果（一个值）。好比这个方法async Task<int> DoAnythingAsync()，DoAnythingAsync的结果是一个整数（一个值）。

因为存在这个限制，你不能将这个功能与yield关键字一块儿使用，而且也不能将其与async IEnumerable<int>（返回异步枚举）一块儿使用。

若是能够将Async/Await特性与yield操做符一块儿使用，咱们就可使用很是强大的编程模型（如异步数据拉取或基于拉取的枚举，在F#中被称为异步序列）。

C# 8中新提出的Async Streams去掉了标量结果的限制，并容许异步方法返回多个结果。

这个变动将使异步模式变得更加灵活，这样就能够按照延迟异步序列的方式从数据库中获取数据，或者按照异步序列的方式下载数据（这些数据在可用时以块的形式返回）。

例如：

foreach await (var streamChunck in asyncStreams) { Console.WriteLine($“Received data count = {streamChunck.Count}”); } 

Reactive Extensions（Rx）是解决异步编程问题的另外一种方法。Rx愈来愈受到开发人员的欢迎。不少其余编程语言（如Java和JavaScript）已经实现了这种技术（RxJava、RxJS）。Rx基于推送式编程模型（Push Programming Model），也称为反应式编程。反应式编程是事件驱动编程的一种类型，它处理的是数据而不是通知。

一般，在推送式编程模型中，你不须要控制Publisher。数据被异步推送到队列中，消费者在数据到达时消费数据。与Rx不一样，Async Streams能够按需被调用，并生成多个值，直到达到枚举的末尾。

在本文中，我将对拉取模型和推送模型进行比较，并演示每一种技术各自的适用场景。我将使用不少代码示例向你展现整个概念和它们的优势，最后，我将讨论Async Streams功能，并向你展现示例代码。

拉取式编程模型与推送式编程模型

图-1-拉取式编程模型与推送式编程模型

我使用的例子是著名的生产者和消费者问题，但在咱们的场景中，生产者不是生成食物，而是生成数据，消费者消费的是生成的数据，如图-1所示。拉取模型很容易理解。消费者询问并拉取生产者的数据。另外一种方法是使用推送模型。生产者将数据发布到队列中，消费者经过订阅队列来接收所需的数据。

拉取模型更合适“快生产者和慢消费者”的场景，由于消费者能够从生产者那里拉取其所需的数据，避免消费者出现溢出。推送模型更适合“慢生产者和快消费者”的场景，由于生产者能够将数据推送给消费者，避免消费者没必要要的等待时间。

Rx和Akka Streams（流式编程模型）使用了回压技术（一种流量控制机制）。它使用拉取模型或推送模型来解决上面提到的生产者和消费者问题。

在下面的示例中，我使用了一个慢消费者从快生产者那里异步拉取数据序列。消费者在处理完一个元素后，会向生产者请求下一个元素，依此类推，直到到达序列的末尾。

动机和背景

要了解咱们为何须要Async Streams，让咱们来看下面的代码。

// 对参数(count)进行循环相加操做 static int SumFromOneToCount(int count) { ConsoleExt.WriteLine("SumFromOneToCount called!"); var sum = 0; for (var i = 0; i <= count; i++) { sum = sum + i; } return sum; }

方法调用：

const int count = 5; ConsoleExt.WriteLine($"Starting the application with count: {count}!"); ConsoleExt.WriteLine("Classic sum starting."); ConsoleExt.WriteLine($"Classic sum result: {SumFromOneToCount(count)}"); ConsoleExt.WriteLine("Classic sum completed."); ConsoleExt.WriteLine("################################################"); ConsoleExt.WriteLine(Environment.NewLine);

输出：

咱们能够经过使用yield运算符让这个方法变成惰性的，以下所示。

static IEnumerable<int> SumFromOneToCountYield(int count) { ConsoleExt.WriteLine("SumFromOneToCountYield called!"); var sum = 0; for (var i = 0; i <= count; i++) { sum = sum + i; yield return sum; } }

调用方法：

const int count = 5; ConsoleExt.WriteLine("Sum with yield starting."); foreach (var i in SumFromOneToCountYield(count)) { ConsoleExt.WriteLine($"Yield sum: {i}"); } ConsoleExt.WriteLine("Sum with yield completed."); ConsoleExt.WriteLine("################################################"); ConsoleExt.WriteLine(Environment.NewLine);

输出：

正如你在输出窗口中看到的那样，结果被分红几个部分返回，而不是做为一个值返回。以上显示的累积结果被称为惰性枚举。可是，仍然存在一个问题，即sum方法阻塞了代码的执行。若是你查看线程，能够看到全部东西都在主线程中运行。

如今，让咱们将async应用于第一个方法SumFromOneToCount上（没有yield关键字）。

static async Task<int> SumFromOneToCountAsync(int count) { ConsoleExt.WriteLine("SumFromOneToCountAsync called!"); var result = await Task.Run(() => { var sum = 0; for (var i = 0; i <= count; i++) { sum = sum + i; } return sum; }); return result; }

调用方法：

const int count = 5; ConsoleExt.WriteLine("async example starting."); // 相加操做是异步进行得！这样还不够，咱们要求不只是异步的，还必须是惰性的。 var result = await SumFromOneToCountAsync(count); ConsoleExt.WriteLine("async Result: " + result); ConsoleExt.WriteLine("async completed."); ConsoleExt.WriteLine("################################################"); ConsoleExt.WriteLine(Environment.NewLine);

输出：

咱们能够看到计算过程是在另外一个线程中运行，但结果仍然是做为一个值返回！

想象一下，咱们能够按照命令式风格将惰性枚举（yield return）与异步方法结合起来。这种组合称为Async Streams。这是C# 8中新提出的功能。这个新功能为咱们提供了一种很好的技术来解决拉取式编程模型问题，例如从网站下载数据或从文件或数据库中读取记录。

让咱们尝试使用当前的C# 版本。我将async关键字添加到SumFromOneToCountYield方法中，以下所示。

图-2 组合使用async关键字和yield发生错误

咱们试着将async添加到SumFromOneToCountYield，但直接出现错误，如上所示！

让咱们试试别的吧。咱们能够将IEnumerable放入任务中并删除yield关键字，以下所示：

static async Task<IEnumerable<int>> SumFromOneToCountTaskIEnumerable(int count) { ConsoleExt.WriteLine("SumFromOneToCountAsyncIEnumerable called!"); var collection = new Collection<int>(); var result = await Task.Run(() => { var sum = 0; for (var i = 0; i <= count; i++) { sum = sum + i; collection.Add(sum); } return collection; }); return result; }

调用方法：

const int count = 5; ConsoleExt.WriteLine("SumFromOneToCountAsyncIEnumerable started!"); var scs = await SumFromOneToCountTaskIEnumerable(count); ConsoleExt.WriteLine("SumFromOneToCountAsyncIEnumerable done!"); foreach (var sc in scs) { // 这不是咱们想要的，结果将做为块返回!!!! ConsoleExt.WriteLine($"AsyncIEnumerable Result: {sc}"); } ConsoleExt.WriteLine("################################################"); ConsoleExt.WriteLine(Environment.NewLine);

输出：

能够看到，咱们异步计算全部的内容，但仍然存在一个问题。结果（全部结果都在集合中累积）做为一个块返回，但这不是咱们想要的惰性行为，咱们的目标是将惰性行为与异步计算风格相结合。

为了实现所需的行为，你须要使用外部库，如Ix（Rx的一部分），或者你必须使用新提出的C#特性Async Streams。

回到咱们的代码示例。我使用了一个外部库来显示异步行为。

static async Task ConsumeAsyncSumSeqeunc(IAsyncEnumerable<int> sequence) { ConsoleExt.WriteLineAsync("ConsumeAsyncSumSeqeunc Called"); await sequence.ForEachAsync(value => { ConsoleExt.WriteLineAsync($"Consuming the value: {value}"); // 模拟延迟! Task.Delay(TimeSpan.FromSeconds(1)).Wait(); }); } static IEnumerable<int> ProduceAsyncSumSeqeunc(int count) { ConsoleExt.WriteLineAsync("ProduceAsyncSumSeqeunc Called"); var sum = 0; for (var i = 0; i <= count; i++) { sum = sum + i; // 模拟延迟! Task.Delay(TimeSpan.FromSeconds(0.5)).Wait(); yield return sum; } }

调用方法：

const int count = 5; ConsoleExt.WriteLine("Starting Async Streams Demo!"); // 启动一个新任务，用于生成异步数据序列! IAsyncEnumerable<int> pullBasedAsyncSequence = ProduceAsyncSumSeqeunc(count).ToAsyncEnumerable(); ConsoleExt.WriteLineAsync("X#X#X#X#X#X#X#X#X#X# Doing some other work X#X#X#X#X#X#X#X#X#X#"); // 启动另外一个新任务，用于消费异步数据序列! var consumingTask = Task.Run(() => ConsumeAsyncSumSeqeunc(pullBasedAsyncSequence)); // 出于演示目的，等待任务完成! consumingTask.Wait(); ConsoleExt.WriteLineAsync("Async Streams Demo Done!");

输出：

最后，咱们实现了咱们想要的行为！咱们能够在枚举上进行异步迭代。

源代码在这里。

客户端/服务器端的异步拉取

我将使用一个更现实的例子来解释这个概念。客户端/服务器端架构是演示这一功能优点的绝佳方法。

客户端/服务器端同步调用

客户端向服务器端发送请求，客户端必须等待（客户端被阻塞），直到服务器端作出响应，如图-3所示。

图-3 同步数据拉取，客户端等待请求完成

异步数据拉取

客户端发出数据请求而后继续执行其余操做。一旦有数据到达，客户端就继续处理达到的数据。

图-4 异步数据拉取，客户端能够在请求数据时执行其余操做

异步序列数据拉取

客户端发出数据块请求，而后继续执行其余操做。一旦数据块到达，客户端就处理接收到的数据块并询问下一个数据块，依此类推，直到达到最后一个数据块为止。这正是Async Streams想法的来源。图-5显示了客户端能够在收到任何数据时执行其余操做或处理数据块。

图-5 异步序列数据拉取（Async Streams），客户端未被阻塞！

Async Streams

与IEnumerable<T>和IEnumerator<T>相似，Async Streams提供了两个新接口IAsyncEnumerable<T>和IAsyncEnumerator<T>，定义以下：

public interface IAsyncEnumerable<out T> { IAsyncEnumerator<T> GetAsyncEnumerator(); } public interface IAsyncEnumerator<out T> : IAsyncDisposable { Task<bool> MoveNextAsync(); T Current { get; } } // Async Streams Feature能够被异步销毁 public interface IAsyncDisposable { Task DiskposeAsync(); }

Jonathan Allen已经在InfoQ网站上介绍过这个主题，我不想在这里再重复一遍，因此我建议你也阅读一下他的文章。

关键在于Task<bool> MoveNextAsync()的返回值（从bool改成Task<bool>，bool IEnumerator.MoveNext()）。这样可让整个计算和迭代都保持异步。大多数状况下，这仍然是拉取模型，即便它是异步的。IAsyncDisposable接口可用于进行异步清理。有关异步的更多信息，请点击此处。

语法

最终语法应以下所示：

foreach await (var dataChunk in asyncStreams) { // 处理数据块或作一些其余的事情! }

如上所示，咱们如今能够按顺序计算多个值，而不仅是计算单个值，同时还可以等待其余异步操做结束。

重写微软的示例

我重写了微软的演示代码，你能够从个人GitHub下载相关代码。

这个例子背后的想法是建立一个大的MemoryStream（20000字节的数组），并按顺序异步迭代集合中的元素或MemoryStream。每次迭代从数组中拉取8K字节。

在(1)处，咱们建立了一个大字节数组并填充了一些虚拟值。在(2)处，咱们定义了一个叫做checksum的变量。咱们将使用checksum来确保计算的总和是正确的。数组和checksum位于内存中，并经过一个元组返回，如(3)所示。

在(4)处，AsEnumarble（或者叫AsAsyncEnumarble）是一种扩展方法，用于模拟由8KB块组成的异步流（ (6)处所示的BufferSize = 8000）。

一般，你没必要继承IAsyncEnumerable，但在上面的示例中，微软这样作是为了简化演示，如(5)处所示。

(7)处是“foreach”，它从异步内存流中拉取8KB的块数据。当消费者（foreach代码块）准备好接收更多数据时，拉取过程是顺序进行的，而后它从生产者（内存流数组）中拉取更多的数据。最后，当迭代完成后，应用程序将’c’的校验和与checksum进行比较，若是它们匹配，就打印出“Checksums match！”，如(8)所示！

微软演示的输出窗口：

概要

咱们已经讨论过Async Streams，它是一种出色的异步拉取技术，可用于进行生成多个值的异步计算。

Async Streams背后的编程概念是异步拉取模型。咱们请求获取序列的下一个元素，并最终获得答复。这与IObservable<T>的推送模型不一样，后者生成与消费者状态无关的值。Async Streams提供了一种表示异步数据源的绝佳方法，例如，当消费者还没有准备好处理更多数据时。示例包含了Web应用程序或从数据库中读取记录。

原文连接：http://www.infoq.com/cn/articles/Async-Streams