多线程调用Azure Function写入Blob递增序号的线程安全实现咨询

核心解决多并发下的读写竞态问题，直接使用「读Blob取值+1写回」的逻辑会出现并发覆盖导致计数重复，这里采用Azure Blob原生的乐观并发控制（ETag校验） 实现原子更新，无需引入额外的中间件即可保证计数严格递增不重复。

首先需要安装NuGet包：Azure.Storage.Blobs

using Azure;
using Azure.Storage.Blobs;
using Azure.Storage.Blobs.Models;
using Microsoft.AspNetCore.Http;
using Microsoft.AspNetCore.Mvc;
using Microsoft.Azure.WebJobs;
using Microsoft.Azure.WebJobs.Extensions.Http;
using Microsoft.Extensions.Logging;
using System.IO;
using System.Threading.Tasks;

public static class SequenceIncrementFunction
{
    // Blob存储连接字符串，建议存在Function的应用配置中，不要硬编码
    private const string BlobConnectionString = "YOUR_BLOB_CONNECTION_STRING";
    private const string ContainerName = "sequence-container";
    private const string BlobName = "current-sequence.txt";

    [FunctionName("IncrementSequence")]
    public static async Task<IActionResult> Run(
        [HttpTrigger(AuthorizationLevel.Function, "get", "post", Route = null)] HttpRequest req,
        ILogger log)
    {
        var blobServiceClient = new BlobServiceClient(BlobConnectionString);
        var containerClient = blobServiceClient.GetBlobContainerClient(ContainerName);
        await containerClient.CreateIfNotExistsAsync();
        var blobClient = containerClient.GetBlobClient(BlobName);

        int newSequence = 0;
        // 循环重试直到写入成功
        while (true)
        {
            try
            {
                int currentSequence;
                ETag currentEtag;

                // 情况1：Blob已经存在，读取当前值和ETag
                if (await blobClient.ExistsAsync())
                {
                    var downloadResponse = await blobClient.DownloadAsync();
                    using var streamReader = new StreamReader(downloadResponse.Value.Content);
                    var content = await streamReader.ReadToEndAsync();
                    currentSequence = int.TryParse(content, out var num) ? num : 0;
                    currentEtag = downloadResponse.Value.Details.ETag;
                    
                    newSequence = currentSequence + 1;
                }
                // 情况2：Blob不存在，首次调用初始值为1
                else
                {
                    newSequence = 1;
                    // 首次写入要求Blob不存在才允许写入，避免并发创建冲突
                    await blobClient.UploadAsync(BinaryData.FromString(newSequence.ToString()), 
                        new BlobUploadOptions { Conditions = new BlobRequestConditions { IfNoneMatch = ETag.All } });
                    break;
                }

                // 携带ETag做写入校验，只有当前Blob的ETag和读取到的一致时才允许写入
                await blobClient.UploadAsync(BinaryData.FromString(newSequence.ToString()), 
                    new BlobUploadOptions { Conditions = new BlobRequestConditions { IfMatch = currentEtag } },
                    overwrite: true);
                break;
            }
            catch (RequestFailedException ex) when (ex.Status == 412)
            {
                // 412错误代表ETag不匹配，有其他请求已经更新了Blob，重试即可
                log.LogInformation("并发冲突，重试更新计数");
            }
        }

        return new OkObjectResult(new { SequenceNumber = newSequence });
    }
}

• 所有并发控制逻辑都在Function侧实现，控制台多线程调用端无需做额外处理，直接正常发起HTTP请求即可
• 每次写入前都会校验ETag，只有当前Blob没有被其他请求修改的情况下写入才会成功，保证不会出现并发覆盖导致的计数重复
• 冲突时自动重试，不会出现漏计数、跳号的问题，最终Blob中的数值等于总调用次数

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