10

我不明白为什么，但客户端库中似乎没有机制可以为 Windows Azure 表存储并行执行许多查询。我创建了一个模板类，可以用来节省大量时间，欢迎您随意使用它。但是，如果您能将其分开，并提供有关如何改进该课程的反馈，我将不胜感激。

public class AsyncDataQuery<T> where T: new()
{
    public AsyncDataQuery(bool preserve_order)
    {
        m_preserve_order = preserve_order;
        this.Queries = new List<CloudTableQuery<T>>(1000);
    }

    public void AddQuery(IQueryable<T> query)
    {
        var data_query = (DataServiceQuery<T>)query;
        var uri = data_query.RequestUri; // required

        this.Queries.Add(new CloudTableQuery<T>(data_query));
    }

    /// <summary>
    /// Blocking but still optimized.
    /// </summary>
    public List<T> Execute()
    {
        this.BeginAsync();
        return this.EndAsync();
    }

    public void BeginAsync()
    {
        if (m_preserve_order == true)
        {
            this.Items = new List<T>(Queries.Count);
            for (var i = 0; i < Queries.Count; i++)
            {
                this.Items.Add(new T());
            }
        }
        else
        {
            this.Items = new List<T>(Queries.Count * 2);
        }

        m_wait = new ManualResetEvent(false);

        for (var i = 0; i < Queries.Count; i++)
        {
            var query = Queries[i];
            query.BeginExecuteSegmented(callback, i);
        }
    }

    public List<T> EndAsync()
    {
        m_wait.WaitOne();
        m_wait.Dispose();

        return this.Items;
    }

    private List<T> Items { get; set; }
    private List<CloudTableQuery<T>> Queries { get; set; }

    private bool m_preserve_order;
    private ManualResetEvent m_wait;
    private int m_completed = 0;
    private object m_lock = new object();

    private void callback(IAsyncResult ar)
    {
        int i = (int)ar.AsyncState;
        CloudTableQuery<T> query = Queries[i];
        var response = query.EndExecuteSegmented(ar);
        if (m_preserve_order == true)
        { // preserve ordering only supports one result per query
            lock (m_lock)
            {
                this.Items[i] = response.Results.Single();
            }
        }
        else
        { // add any number of items
            lock (m_lock)
            {
                this.Items.AddRange(response.Results);
            }
        }
        if (response.HasMoreResults == true)
        { // more data to pull
            query.BeginExecuteSegmented(response.ContinuationToken, callback, i);
            return;
        }
        m_completed = Interlocked.Increment(ref m_completed);
        if (m_completed == Queries.Count)
        {
            m_wait.Set();
        }
    }
}
4

2 回答 2

5

猜猜我参加晚会迟到了。我要补充两点：

  1. ManualResetEvent 是 IDisposable。所以你需要确保它被放置在某个地方。
  2. 错误处理 - 如果其中一个查询失败，它可能会失败整个事情。您可能应该重试失败的请求。或者，您可以返回您确实返回的值，并指示哪些查询失败，以便调用者可以重试查询。
  3. 客户端超时 - 没有。如果服务器端超时，这不是问题，但如果失败（例如，网络问题），客户端将永远挂起。

另外，我认为这实际上是比任务并行库更好的方法。在此之前，我尝试了 Task-per-query 方法。代码实际上更尴尬，而且往往会导致有很多活动线程。我还没有对你的代码进行广泛的测试，但它似乎在第一次脸红时效果更好。

更新

我对上面的代码进行了或多或少的重写。我的重写删除了所有锁定，支持挂起事务的客户端超时（很少见，但确实会发生，并且真的会毁了你的一天），以及一些异常处理逻辑。有一个完整的解决方案，在Bitbucket上进行了测试。最相关的代码存在于一个文件中，尽管它确实需要项目其他部分中的一些帮助程序。

于 2012-02-18T00:47:07.097 回答
4

您是否考虑过使用任务并行库？

http://msdn.microsoft.com/en-us/library/dd537609.aspx

于 2010-12-29T22:28:30.963 回答