如何规避Azure Function更新Azure Table Storage指标表的竞态条件

嘿，我知道你现在被多实例并发更新metrics表的竞态问题困扰，咱们先来揪出现有代码里的小bug，再一步步解决核心问题。

先修复现有重试逻辑的细节问题 #

你的递归重试里藏着一个容易忽略的坑：retryDepth++是后自增，这意味着传递给下一次递归的还是原来的retryDepth值，导致重试次数根本达不到预期的3次。得改成retryDepth + 1，确保每次重试深度正确递增：

if (retryDepth < maxRetryDepth)
    await UpdateMetricEntry(auditTableService, sourceSystemReference, addNewBytes, addIncrementBytes, retryDepth + 1);

另外，建议把DateTime.Now换成DateTime.UtcNow来生成年月标识，避免不同实例的本地时区/时间差异导致行键不一致，比如：

var todayYearMonth = DateTime.UtcNow.ToString("yyyyMM");

核心问题：乐观并发的局限性与进阶方案 #

乐观并发（ETag）+重试是Table Storage的标准方案，但6个实例同时更新同一条记录时，冲突概率会显著上升，单纯重试可能还是会有部分更新丢失。下面是几个更可靠的解决方案：

1. 优化乐观并发的重试策略 #

用专业的重试库替代递归，比如Polly，实现指数退避重试，既能降低冲突概率，也更易维护：

// 定义重试策略：针对412错误，最多重试3次，每次等待时间指数递增
var retryPolicy = Policy
    .Handle<StorageException>(ex => ex.RequestInformation.HttpStatusCode == 412)
    .WaitAndRetryAsync(3, retryAttempt => 
        TimeSpan.FromMilliseconds(Math.Pow(2, retryAttempt) * 100));

await retryPolicy.ExecuteAsync(async () =>
{
    var todayYearMonth = DateTime.UtcNow.ToString("yyyyMM");
    var result = await auditTableService.GetRecord<VolumeMetric>("VolumeMetrics", sourceSystemReference, todayYearMonth);
    
    VolumeMetric volumeMetric = result.RecordExists 
        ? (VolumeMetric)result.Record.Clone() 
        : new VolumeMetric {
            PartitionKey = sourceSystemReference,
            RowKey = todayYearMonth,
            SourceSystemReference = sourceSystemReference,
            BillingMonth = DateTime.UtcNow.Month,
            BillingYear = DateTime.UtcNow.Year,
            ETag = "*"
        };
    
    volumeMetric.NewVolumeBytes += addNewBytes;
    volumeMetric.IncrementalVolumeBytes += addIncrementBytes;
    await auditTableService.InsertOrReplace("VolumeMetrics", volumeMetric);
});

2. 引入分布式锁彻底避免并发 #

如果乐观并发的冲突率还是太高，可以用Azure Redis Cache实现分布式锁，确保同一时间只有一个实例能更新特定的metrics记录：

// 假设已初始化Redis连接
var redis = ConnectionMultiplexer.Connect("your-redis-connection-string");
var db = redis.GetDatabase();
var lockKey = $"metric-lock:{sourceSystemReference}:{DateTime.UtcNow.ToString("yyyyMM")}";

// 获取锁，设置10秒过期（防止实例崩溃导致死锁）
var lockAcquired = await db.LockTakeAsync(lockKey, Environment.MachineName, TimeSpan.FromSeconds(10));
if (!lockAcquired)
{
    // 没拿到锁，短暂等待后重试
    await Task.Delay(TimeSpan.FromMilliseconds(500));
    await UpdateMetricEntry(auditTableService, sourceSystemReference, addNewBytes, addIncrementBytes, retryDepth + 1);
    return;
}

try
{
    // 执行metrics更新逻辑（和之前的代码一致）
    var todayYearMonth = DateTime.UtcNow.ToString("yyyyMM");
    var result = await auditTableService.GetRecord<VolumeMetric>("VolumeMetrics", sourceSystemReference, todayYearMonth);
    // ... 构造实体并更新
}
finally
{
    // 释放锁
    await db.LockReleaseAsync(lockKey, Environment.MachineName);
}

这种方式能彻底消除并发冲突，但需要额外部署Redis，还要处理锁过期、释放失败等边缘情况。

3. 改用Azure Cosmos DB for Table API #

如果你的架构允许切换存储服务，Cosmos DB的Table API是更优选择：

• 支持服务器端存储过程，可以在服务器原子性完成“查询-修改-更新”操作，完全避免客户端竞态
• 内置更强的重试机制和灵活的一致性级别（比如会话一致性）
• 更高的吞吐量和更低的延迟

比如写一个Cosmos DB存储过程，直接在服务器端累加字节数：

function updateVolumeMetric(sourceSystem, yearMonth, newBytes, incrementalBytes) {
    var collection = getContext().getCollection();
    var query = `SELECT * FROM c WHERE c.PartitionKey = '${sourceSystem}' AND c.RowKey = '${yearMonth}'`;
    
    collection.queryDocuments(collection.getSelfLink(), query, {}, function(err, docs) {
        if (err) throw err;
        
        var metric;
        if (docs.length === 0) {
            // 创建新记录
            metric = {
                PartitionKey: sourceSystem,
                RowKey: yearMonth,
                NewVolumeBytes: newBytes,
                IncrementalVolumeBytes: incrementalBytes
            };
        } else {
            // 更新现有记录
            metric = docs[0];
            metric.NewVolumeBytes += newBytes;
            metric.IncrementalVolumeBytes += incrementalBytes;
        }
        
        collection.upsertDocument(collection.getSelfLink(), metric, {}, function(err) {
            if (err) throw err;
            getContext().getResponse().setBody(metric);
        });
    });
}

然后在Function里调用这个存储过程，就能保证原子性更新。

4. 利用Service Bus会话锁从根源避免并发 #

如果你的消息是按sourceSystemRef分组的，可以给每个消息设置会话ID为sourceSystemRef的值。Service Bus会把同一会话的消息路由到同一个Function实例，这样同一系统的metrics更新只会由一个实例处理，从根源上避免冲突。

在Function的Service Bus触发器配置里开启会话支持：

[FunctionName("ProcessFileInfo")]
public static async Task Run(
    [ServiceBusTrigger("your-topic", "your-subscription", Connection = "ServiceBusConnection", IsSessionsEnabled = true)] string mySbMsg,
    ILogger log)
{
    // 处理消息逻辑
}

发送消息时设置会话ID：

var message = new Message(Encoding.UTF8.GetBytes(jsonPayload))
{
    SessionId = sourceSystemRef
};
await sender.SendAsync(message);

这个方案不需要额外存储服务，但会限制同一sourceSystem的消息只能由一个实例处理，适合sourceSystem数量较多、单个系统消息量不大的场景。

总结 #

如果不想改动现有架构，先修复重试逻辑的bug，再用Polly优化重试策略；如果冲突率还是很高，引入分布式锁是比较直接的方案；如果可以升级存储服务，Cosmos DB的存储过程是最可靠的原子更新方式；最后，Service Bus会话锁适合消息按系统分组的场景。

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