我不明白为什么,但客户端库中似乎没有机制可以并行执行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();
}
}
}
猜猜我迟到了。 我会添加两件事:
另外,我认为这实际上是一个比任务并行库更好的方法。 在此之前我尝试了每查询任务的方法。 代码实际上更尴尬,它往往导致有很多活动线程。 我仍然没有对你的代码进行过广泛的测试,但它在第一次脸红时效果似乎更好。
我已经将一些工作变成了或多或少的重写代码。 我的重写删除了所有锁定,支持挂起事务的客户端超时(很少见,但它确实发生了,并且可能真的毁了你的一天),以及一些异常处理逻辑。 在Bitbucket上有一个完整的测试解决方案。 最相关的代码存在于一个文件中 ,尽管它确实需要项目其他部分中的一些帮助程序。
您是否考虑过使用任务并行库?