如何处理Azure Durable Functions中Activity Function重复调用问题？

针对你提到的CopyFileToBlob这类Activity函数重复调用的问题，核心是通过业务状态校验+分布式锁实现幂等，确保多次调用只执行一次有效操作，同时复用已有结果。以下是具体落地方案：

核心思路 #

Activity的幂等性不能依赖Durable Functions的内置机制，必须结合业务场景做两层保障：

1. 结果校验：先判断操作是否已完成（比如目标Blob是否存在），直接复用结果；
2. 分布式锁：防止并发调用时的竞态问题，确保同一时间只有一个实例执行操作。

改造CopyFileToBlob函数示例（C#） #

以官方示例中的文件复制场景为例，改造后代码如下：

[FunctionName("CopyFileToBlob")]
public static async Task<string> Run(
    [ActivityTrigger] FileCopyRequest request,
    [BlobServiceClient] BlobServiceClient blobServiceClient,
    ILogger log)
{
    var targetContainer = blobServiceClient.GetBlobContainerClient(request.TargetContainer);
    await targetContainer.CreateIfNotExistsAsync();
    var targetBlob = targetContainer.GetBlobClient(request.TargetBlobPath);

    // 第一步：先检查目标是否已存在，直接返回结果（避免重复执行）
    if (await targetBlob.ExistsAsync())
    {
        log.LogInformation($"Blob {request.TargetBlobPath} 已存在，跳过复制");
        return targetBlob.Uri.ToString();
    }

    // 第二步：用Blob租约做分布式锁，防止并发调用
    var leaseClient = targetBlob.GetBlobLeaseClient();
    try
    {
        // 获取15秒租约（根据实际复制耗时调整时长）
        await leaseClient.AcquireAsync(TimeSpan.FromSeconds(15));

        // 再次检查目标状态（防止获取租约前已有其他实例完成操作）
        if (await targetBlob.ExistsAsync())
        {
            log.LogInformation($"获取租约期间Blob {request.TargetBlobPath} 已创建，跳过复制");
            return targetBlob.Uri.ToString();
        }

        // 执行实际复制逻辑
        var sourceBlob = blobServiceClient.GetBlobContainerClient(request.SourceContainer)
                                          .GetBlobClient(request.SourceBlobPath);
        await targetBlob.StartCopyFromUriAsync(sourceBlob.Uri);

        // 等待复制完成（可选，根据业务是否需要同步返回结果）
        var copyProps = await targetBlob.GetPropertiesAsync();
        while (copyProps.Value.CopyStatus == CopyStatus.Pending)
        {
            await Task.Delay(1000);
            copyProps = await targetBlob.GetPropertiesAsync();
        }

        if (copyProps.Value.CopyStatus != CopyStatus.Success)
        {
            throw new Exception($"复制失败，状态：{copyProps.Value.CopyStatus}");
        }

        log.LogInformation($"成功复制 {request.SourceBlobPath} 到 {request.TargetBlobPath}");
        return targetBlob.Uri.ToString();
    }
    catch (RequestFailedException ex) when (ex.ErrorCode == BlobErrorCode.LeaseAlreadyPresent)
    {
        // 租约被占用，说明有其他实例正在执行，等待后检查结果
        log.LogInformation($"Blob {request.TargetBlobPath} 的租约已被其他实例持有，等待结果...");
        await Task.Delay(2000);

        // 再次检查目标是否存在，存在则返回结果
        if (await targetBlob.ExistsAsync())
        {
            return targetBlob.Uri.ToString();
        }

        // 若仍不存在，说明前一个实例执行失败，可根据业务选择重试或抛出异常
        throw new Exception("并发复制操作失败，目标Blob未创建");
    }
    finally
    {
        // 释放租约（忽略租约不存在的情况，比如已过期）
        try
        {
            await leaseClient.ReleaseAsync();
        }
        catch (RequestFailedException ex) when (ex.ErrorCode == BlobErrorCode.LeaseNotFound)
        {
            // 无需处理
        }
    }
}

// 定义请求参数类
public class FileCopyRequest
{
    public string SourceContainer { get; set; }
    public string SourceBlobPath { get; set; }
    public string TargetContainer { get; set; }
    public string TargetBlobPath { get; set; }
}

关键细节说明 #

• 前置检查+二次校验：两次检查目标状态是为了避免竞态——比如第一个实例还没完成操作，第二个实例直接跳过的情况；
• 分布式锁选择：这里用Blob租约是因为场景本身依赖Blob存储，无需额外引入其他服务；如果是其他场景，也可以用Redis锁、Cosmos DB乐观锁等；
• 锁冲突处理：当租约被占用时，不要直接报错，而是等待后检查结果，这样后续调用可以复用第一个实例的执行结果；
• 租约时长：要根据操作的实际耗时设置，确保在操作完成前租约不会过期，避免锁提前释放导致的并发问题。

其他通用幂等方案 #

如果你的Activity不是Blob操作，还可以用以下方式：

• 状态记录表：用Azure Table或Cosmos DB记录每个请求的执行状态（比如用请求参数的哈希值作为唯一键），执行前先查状态，已完成则返回结果，未完成则加锁执行；
• 操作本身幂等：比如数据库操作使用UPSERT而非INSERT，确保重复执行不会产生重复数据。

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