将重要代码转换为新的 .NET 异步模式 - 如何处理 Yield 循环

Question

我正在编写一个与 Azure 表存储一起使用的库。基本模式是给定的 HTTP 请求在内容流中返回多个结果，并在标头中返回指向下一组结果的指针。当从流中读取结果时，就会产生结果。我正在使用 System.Net.Http 库（以前是 Microsoft.Net.Http），它在最新版本中删除了 HttpClient.Send 的同步版本和其他同步方法。新版本使用任务。我以前使用过任务，但不是用于如此复杂的事情，而且我很难开始。

已转换为异步模式的调用有：HttpClient.Send、response.Context.ContentReadSteam。我已经清理了代码，以便显示重要的部分。

var queryUri = _GetTableQueryUri(tableServiceUri, tableName, query, null, null, timeout);
while(true) {
    var continuationParitionKey = "";
    var continuationRowKey = "";
    using (var request = GetRequest(queryUri, null, action.Method, azureAccountName, azureAccountKey))
    {
        using (var client = new HttpClient())
        {
            using (var response = client.Send(request, HttpCompletionOption.ResponseHeadersRead))
            {
                continuationParitionKey = // stuff from headers
                continuationRowKey = // stuff from headers

                using (var reader = XmlReader.Create(response.Content.ContentReadStream))
                {
                    while (reader.Read())
                    {
                        if (reader.NodeType == XmlNodeType.Element && reader.Name == "entry" && reader.NamespaceURI == "http://www.w3.org/2005/Atom")
                        {
                            yield return XElement.ReadFrom(reader) as XElement;
                        }
                    }
                    reader.Close();
                }
            }
        }
    }
    if (continuationParitionKey == null && continuationRowKey == null)
        break;

    queryUri = _GetTableQueryUri(tableServiceUri, tableName, query, continuationParitionKey, continuationRowKey, timeout);
}

下面是我已成功转换的一个示例。

client.SendAsync(request, HttpCompletionOption.ResponseHeadersRead).ContinueWith(task =>
    {
        using (var response = task.Result)
        {
            if (response.StatusCode == HttpStatusCode.Created && action == HttpMethod.Post)
            {
                return XElement.Load(response.Content.ReadAsStreamAsync().Result);
            }
        }
    });

有人对如何将循环/产量转换为新模式有任何建议吗？

谢谢！ 埃里克

Answer 1

正如您所发现的，async 目前并不能与 yield 配合使用。尽管它们进行类似的代码转换，但目标却截然不同。

有两种解决方案：一种是提供缓冲区并使用生产者/消费者类型的方法。 System.Tasks.Dataflow.dll 对于管理复杂代码中的缓冲区非常有用。

另一个解决方案是编写一个“异步枚举器”。这在概念上更接近于您的代码应该执行的操作，但此解决方案比生产者/消费者解决方案复杂得多。

“异步枚举器”类型在 Rx 上的此视频，您可以从 一个实验性的 Rx 包（请注意，尽管这是由 Rx 团队完成的，但它实际上并没有使用 Rx）。

Answer 2

我建议将 Loop\yield 转换为某种形式的输出队列，例如 本文使用 BlockingCollection。因此，方法的调用者为您提供了一个队列来将结果推送到其中。

队列很方便，因为它将生产者与消费者解耦，但这只是一种选择。更一般地说，调用者为您提供一个回调，以便为您获取的每个结果执行。该回调应该是异步的，例如，它可以启动另一个任务。