Azure 表存储的 ExecuteAsync() 未能插入所有记录

Question

我正在尝试将 10000 条记录插入 Azure 表存储。我正在使用 ExecuteAsync() 来实现它，但不知何故，大约插入了大约 7500 条记录，其余记录丢失了。我故意不使用 await 关键字，因为我不想等待结果，只想将它们存储在表中。下面是我的代码 sn-p。

private static async void ConfigureAzureStorageTable()
    {
        CloudStorageAccount storageAccount =
            CloudStorageAccount.Parse(CloudConfigurationManager.GetSetting("StorageConnectionString"));
        CloudTableClient tableClient = storageAccount.CreateCloudTableClient();
        TableResult result = new TableResult();
        CloudTable table = tableClient.GetTableReference("test");
        table.CreateIfNotExists();

        for (int i = 0; i < 10000; i++)
        {
            var verifyVariableEntityObject = new VerifyVariableEntity()
            {
                ConsumerId = String.Format("{0}", i),
                Score = String.Format("{0}", i * 2 + 2),
                PartitionKey = String.Format("{0}", i),
                RowKey = String.Format("{0}", i * 2 + 2)
            };
            TableOperation insertOperation = TableOperation.Insert(verifyVariableEntityObject);
            try
            {
                table.ExecuteAsync(insertOperation);
            }
            catch (Exception e)
            {

                Console.WriteLine(e.Message);
            }
        }
    }

方法的使用有什么不对吗？

Answer 1

你仍然想await table.ExecuteAsync()。这意味着 ConfigureAzureStorageTable() 在此时将控制权返回给调用者，调用者可以继续执行。

按照您在问题中的方式，ConfigureAzureStorageTable() 将继续调用 table.ExecuteAsync() 并退出，而 table 之类的内容将超出范围，而 table.ExecuteAsync() 任务仍然没有完成。

在 SO 和其他地方使用 async void 有很多注意事项，您还需要考虑这些注意事项。您可以像async Task 一样轻松地使用您的方法，但不要在调用者中等待它尚未，而是保留返回的Task 以进行干净的终止等。

编辑：一个补充——你几乎肯定想在你的await 上使用ConfigureAwait(false)，因为你似乎不需要保留任何上下文。这个blog post 有一些关于这方面的指导方针和一般的异步。

Answer 2

根据您的要求，我已成功使用CloudTable.ExecuteAsync 和CloudTable.ExecuteBatchAsync 测试了您的场景。这是我的代码sn-p关于使用CloudTable.ExecuteBatchAsync向Azure表存储插入记录，你可以参考一下。

Program.cs 主目录

class Program
{
    static void Main(string[] args)
    {
        CloudStorageAccount storageAccount =
            CloudStorageAccount.Parse(CloudConfigurationManager.GetSetting("StorageConnectionString"));
        CloudTableClient tableClient = storageAccount.CreateCloudTableClient();
        TableResult result = new TableResult();
        CloudTable table = tableClient.GetTableReference("test");
        table.CreateIfNotExists();

        //Generate records to be inserted into Azure Table Storage
        var entities = Enumerable.Range(1, 10000).Select(i => new VerifyVariableEntity()
        {
            ConsumerId = String.Format("{0}", i),
            Score = String.Format("{0}", i * 2 + 2),
            PartitionKey = String.Format("{0}", i),
            RowKey = String.Format("{0}", i * 2 + 2)
        });

        //Group records by PartitionKey and prepare for executing batch operations
        var batches = TableBatchHelper<VerifyVariableEntity>.GetBatches(entities);

        //Execute batch operations in parallel
        Parallel.ForEach(batches, new ParallelOptions()
        {
            MaxDegreeOfParallelism = 5
        }, (batchOperation) =>
        {
            try
            {
                table.ExecuteBatch(batchOperation);
                Console.WriteLine("Writing {0} records", batchOperation.Count);
            }
            catch (Exception ex)
            {
                Console.WriteLine("ExecuteBatch throw a exception:" + ex.Message);
            }
        });
        Console.WriteLine("Done!");
        Console.WriteLine("Press any key to exit...");
        Console.ReadKey();
    }
}

TableBatchHelper.cs

public class TableBatchHelper<T> where T : ITableEntity
{
    const int batchMaxSize = 100;

    public static IEnumerable<TableBatchOperation> GetBatches(IEnumerable<T> items)
    {
        var list = new List<TableBatchOperation>();
        var partitionGroups = items.GroupBy(arg => arg.PartitionKey).ToArray();
        foreach (var group in partitionGroups)
        {
            T[] groupList = group.ToArray();
            int offSet = batchMaxSize;
            T[] entities = groupList.Take(offSet).ToArray();
            while (entities.Any())
            {
                var tableBatchOperation = new TableBatchOperation();
                foreach (var entity in entities)
                {
                    tableBatchOperation.Add(TableOperation.InsertOrReplace(entity));
                }
                list.Add(tableBatchOperation);
                entities = groupList.Skip(offSet).Take(batchMaxSize).ToArray();
                offSet += batchMaxSize;
            }
        }
        return list;
    }
}

注意：正如官方document关于插入一批实体所说：

单个批处理操作最多可以包含 100 个实体。

单个批处理操作中的所有实体必须具有相同的分区键。

总之，请尝试检查它是否适用于您。此外，您可以在控制台应用程序中捕获详细的异常，并通过Fiddler 捕获 HTTP 请求，以便在将记录插入 Azure 表存储时捕获 HTTP 错误请求。

Answer 3

如何使用 TableBatchOperation 一次运行 N 个批次的插入？

private const int BatchSize = 100;

private static async void ConfigureAzureStorageTable()
{
    CloudStorageAccount storageAccount =
        CloudStorageAccount.Parse(CloudConfigurationManager.GetSetting("StorageConnectionString"));
    CloudTableClient tableClient = storageAccount.CreateCloudTableClient();
    TableResult result = new TableResult();
    CloudTable table = tableClient.GetTableReference("test");
    table.CreateIfNotExists();

    var batchOperation = new TableBatchOperation();

    for (int i = 0; i < 10000; i++)
    {
        var verifyVariableEntityObject = new VerifyVariableEntity()
        {
            ConsumerId = String.Format("{0}", i),
            Score = String.Format("{0}", i * 2 + 2),
            PartitionKey = String.Format("{0}", i),
            RowKey = String.Format("{0}", i * 2 + 2)
        };
        TableOperation insertOperation = TableOperation.Insert(verifyVariableEntityObject);
        batchOperation.Add(insertOperation);

        if (batchOperation.Count >= BatchSize)
        {
            try
            {
                await table.ExecuteBatchAsync(batchOperation);
                batchOperation = new TableBatchOperation();
            }
            catch (Exception e)
            {
                Console.WriteLine(e.Message);
            }
        }
    }

    if(batchOperation.Count > 0)
    {
        try
        {
            await table.ExecuteBatchAsync(batchOperation);
        }
        catch (Exception e)
        {
            Console.WriteLine(e.Message);
        }
    }
}

您可以根据需要调整 BatchSize。小免责声明：我没有尝试运行它，虽然它应该可以工作。

但我不禁想知道为什么你的函数是async void？这应该保留给无法决定接口的事件处理程序和类似的处理程序。在大多数情况下，您希望返回一个任务。因为现在调用者无法捕获此函数中发生的异常。

Answer 4

async void 不是一个好习惯，除非它是一个事件处理程序。

https://msdn.microsoft.com/en-us/magazine/jj991977.aspx

如果您打算将许多记录插入到 Azure 表存储中，那么批量插入是您的最佳选择。

https://msdn.microsoft.com/en-us/library/azure/microsoft.windowsazure.storage.table.tablebatchoperation.aspx

请记住，每个批次的表操作限制为 100 个。

Answer 5

我遇到了同样的问题并通过 强制 ExecuteAsync 在结果存在之前等待结果..

table.ExecuteAsync(insertOperation).GetAwaiter().GetResult()