Azure Table Storage 如何使用C#实现原子化获取或插入操作

核心实现思路 #

利用Azure Table Storage服务端原生的主键唯一性约束实现原子性，不需要额外引入分布式锁：

• 同一个分区键（PartitionKey）+行键（RowKey）的实体，Insert操作仅允许第一次请求执行成功，后续同主键的Insert请求会被服务端直接拒绝，返回409冲突状态码
• 无需提前查询实体是否存在，直接提交插入请求，插入成功则返回新实体，插入冲突则查询返回已存在的实体即可，完全避免竞态条件
• 所有并发请求最终拿到的都是首次插入的实体，不会出现覆盖写入，也不会生成重复实体

代码示例（C# + Microsoft.Azure.Cosmos.Table 1.0.8） #

using Microsoft.Azure.Cosmos.Table;
using System;
using System.Net;
using System.Threading.Tasks;

public class TableStorageService
{
    private readonly CloudTable _cloudTable;

    public TableStorageService(string storageConnectionString, string tableName)
    {
        var storageAccount = CloudStorageAccount.Parse(storageConnectionString);
        var tableClient = storageAccount.CreateCloudTableClient();
        _cloudTable = tableClient.GetTableReference(tableName);
        _cloudTable.CreateIfNotExists();
    }

    /// <summary>
    /// 原子化查询或创建实体，保留首次写入的值
    /// </summary>
    /// <typeparam name="T">实体类型，继承自TableEntity</typeparam>
    /// <param name="partitionKey">分区键</param>
    /// <param name="rowKey">行键</param>
    /// <param name="createEntityFactory">实体不存在时的构造委托，所有调用方生成的默认值需保持一致</param>
    /// <returns>已存在或新创建的实体</returns>
    /// <exception cref="ArgumentNullException"></exception>
    public async Task<T> GetOrCreateEntityAsync<T>(string partitionKey, string rowKey, Func<T> createEntityFactory) where T : TableEntity, new()
    {
        if (createEntityFactory == null)
            throw new ArgumentNullException(nameof(createEntityFactory));

        // 构造待插入的实体
        T newEntity = createEntityFactory();
        newEntity.PartitionKey = partitionKey;
        newEntity.RowKey = rowKey;

        try
        {
            // 直接执行插入操作，依赖服务端原子性校验
            var insertOperation = TableOperation.Insert(newEntity);
            var result = await _cloudTable.ExecuteAsync(insertOperation);
            return (T)result.Result;
        }
        catch (StorageException ex) when (ex.RequestInformation.HttpStatusCode == (int)HttpStatusCode.Conflict)
        {
            // 捕获409冲突，说明实体已被其他请求插入，查询返回已有实体
            var retrieveOperation = TableOperation.Retrieve<T>(partitionKey, rowKey);
            var retrieveResult = await _cloudTable.ExecuteAsync(retrieveOperation);
            return (T)retrieveResult.Result;
        }
    }
}

// 自定义实体示例
public class MyBusinessEntity : TableEntity
{
    public string BusinessField1 { get; set; }
    public int BusinessField2 { get; set; }
}

// 调用示例
// var service = new TableStorageService("你的存储账户连接字符串", "你的表名");
// var entity = await service.GetOrCreateEntityAsync<MyBusinessEntity>("test_partition", "test_row", () => new MyBusinessEntity
// {
//     BusinessField1 = "默认值",
//     BusinessField2 = 0
// });

注意事项 #

• 不要提前做查询判断：提前查询会引入额外的竞态窗口，直接提交插入请求是性能和原子性最优的方案
• 异常捕获要精准：仅捕获409状态码的存储异常，其他异常（网络错误、权限错误等）需正常抛出，不要吞异常
• 实体默认值要全局一致：所有并发请求的实体构造逻辑生成的默认值必须相同，才能保证不同请求的返回结果完全一致

内容的提问来源于stack exchange，提问作者Claus Appel