有人可以描述这个递归异步 Azure 存储调用中的逻辑流程吗？

Question

有人可以帮助我理解这段代码是如何执行的，“结果”在哪里，以及我什么时候可以开始用完整的结果做事情。

protected void QuerySongsExecuteSegmentedAsync(CloudTableClient cloudTableClient)
{
  TableServiceContext tableServiceContext = cloudTableClient.GetDataServiceContext();
  tableServiceContext.ResolveType = (unused) => typeof(Song); 

  CloudTableQuery<Song> cloudTableQuery =
    (from entity in tableServiceContext.CreateQuery<Song>("Songs").Take(10)
    select entity ).AsTableServiceQuery<Song>();
  IAsyncResult iAsyncResult =
    cloudTableQuery.BeginExecuteSegmented(BeginExecuteSegmentedIsDone, cloudTableQuery);
} 

static void BeginExecuteSegmentedIsDone(IAsyncResult result)
{
  CloudTableQuery<Song> cloudTableQuery = result.AsyncState as CloudTableQuery<Song>;
  ResultSegment<Song> resultSegment = cloudTableQuery.EndExecuteSegmented(result); 

  List<Song> listSongs = resultSegment.Results.ToList<Song>(); 

  if (resultSegment.HasMoreResults)
  {
    IAsyncResult iAsyncResult =
      cloudTableQuery.BeginExecuteSegmented(
       resultSegment.ContinuationToken, BeginExecuteSegmentedIsDone, cloudTableQuery);
  }
}

Answer 1

针对 Windows Azure 表存储的查询可以返回带有继续标记的部分结果，这意味着您需要重新发出查询（带有继续标记）才能获取下一批结果。通常，您会看到仅使用 .AsTableServiceQuery() 然后进行枚举的代码，这将导致该调用链在枚举期间透明地发生。

此代码通过使用 BeginExecuteSegmented 来检索每批结果来显式执行此操作。就在 List行下方listSongs = resultSegment.Results.ToList()，您应该能够使用这些 Song。 （可能还会有更多结果，但这些结果应该是有效且可用的。）

Answer 2

如果您想使用 TPL（任务并行库）异步执行，您可以使用这样的模式。

    public static Task<IEnumerable<T>> ExecuteAsync<T>(DataServiceQuery<T> query, TableServiceContext tableContext)
    {
        List<T> values = null;

        var cloudQuery = query.AsTableServiceQuery();

        Func<ResultSegment<T>, Task<ResultSegment<T>>> getSegment = null;

        getSegment = new Func<ResultSegment<T>, Task<ResultSegment<T>>>((previous) =>
        {
            return ExecuteSegmentAsync(cloudQuery, previous).ContinueWith(exec =>
            {                    
                if (exec.IsFaulted || exec.IsCanceled)
                {
                    return exec;
                }
                else
                {
                    var segment = exec.Result;
                    if(segment.HasMoreResults)
                    {
                        if(values == null)
                        {
                            values = new List<T>();
                        }
                        values.AddRange(segment.Results);
                        return getSegment(segment);                        
                    }
                    else
                    {
                        return exec;
                    }
                }
            }).Unwrap();
        });

        return getSegment(null).ContinueWith(exec =>
        {
            exec.ThrowOnError("ExecuteAsync");
            var segment = exec.Result;

            if(values == null)
            {
                return segment.Results;
            }
            else
            {
                values.AddRange(segment.Results);
                return values;
            }
        });       
    }

    private static Task<ResultSegment<T>> ExecuteSegmentAsync<T>(CloudTableQuery<T> query, ResultSegment<T> previous)
    {
        var tcs = new TaskCompletionSource<ResultSegment<T>>();

        RetryPolicy.ExecuteAction(ac =>
        {
            // Invoke the begin method of the asynchronous call.
            if(previous != null)
            {
                query.BeginExecuteSegmented(previous.ContinuationToken, ac, null);                
            }
            else
            {
                query.BeginExecuteSegmented(ac, null);                
            }
        },
        ar =>
        {
            // Invoke the end method of the asynchronous call.
            return query.EndExecuteSegmented(ar);
        },
        values =>
        {
            tcs.SetResult(values);                
        },
        ex =>
        {
            // Async opetation failed after multiple retries
            tcs.SetException(ex);
        });

        return tcs.Task;
    }