如何确保Azure Function OperationalFunction仅并行执行一次？

核心思路 #

让所有Blob/Http触发的请求统一进入同一个持久化编排实例，由该编排内部维护任务队列，严格按FIFO顺序逐个调用OperationalFunction，从根源上保证同一时间仅存在一个运行实例。

具体实现步骤 #

1. 修改入口函数：复用固定编排实例ID #

原入口函数每次调用StartNewAsync会创建新编排实例，导致多实例并行。现在改为指定固定实例ID，不存在则创建，存在则发送外部事件传递任务数据：

// 定义全局唯一的单例编排ID
const string SingletonOrchestratorId = "OperationalTaskSingleton";

var starter = context.GetDurableClient();
var status = await starter.GetStatusAsync(SingletonOrchestratorId);

// 实例不存在或已完成时，启动新的单例编排
if (status == null || status.RuntimeStatus == OrchestrationRuntimeStatus.Completed)
{
    await starter.StartNewAsync(nameof(OrchestrationFunction), SingletonOrchestratorId, null);
}

// 向单例编排发送任务事件
await starter.RaiseEventAsync(SingletonOrchestratorId, "NewTask", data);

上述代码需同时修改BlobTriggeredStartFunction和HttpTriggeredStartFunction

2. 修改编排器函数：维护任务队列并串行执行 #

编排器监听外部事件收集任务，按顺序逐个调用Activity函数，确保串行执行：

[FunctionName("OrchestrationFunction")]
public async Task OrchestrationFunction(
    [OrchestrationTrigger] IDurableOrchestrationContext context)
{
    var taskQueue = new Queue<string>();
    // 注册外部事件监听器，持续接收新任务
    var eventListener = context.WaitForExternalEvent<string>("NewTask");

    while (true)
    {
        // 同时等待新任务和超时定时器（避免无任务时长时间占用资源）
        var completedTask = await Task.WhenAny(
            eventListener, 
            context.CreateTimer(context.CurrentUtcDateTime.AddHours(24), CancellationToken.None)
        );

        if (completedTask == eventListener)
        {
            // 收到新任务，加入队列
            var taskData = await eventListener;
            taskQueue.Enqueue(taskData);
            // 重新注册监听器，准备接收下一个任务
            eventListener = context.WaitForExternalEvent<string>("NewTask");
        }
        else
        {
            // 超时无新任务，结束编排
            break;
        }

        // 串行执行队列中的所有任务
        while (taskQueue.Count > 0)
        {
            var currentTask = taskQueue.Dequeue();
            await context.CallActivityAsync(nameof(OperationalFunction), currentTask);
        }
    }
}

方案优势 #

• 严格串行执行：所有任务进入同一个编排队列，绝对不会出现OperationalFunction并行实例
• 无轮询开销：依赖Durable Functions的外部事件机制，编排器主动等待任务，资源消耗极低
• 全场景覆盖：同时支持Blob和Http触发的请求，无遗漏
• 持久化可靠：队列和编排状态由Durable框架持久化，函数重启或异常时任务不会丢失

替代方案：分布式锁（Azure Redis Cache） #

若不想修改编排逻辑，可在OperationalFunction内部加分布式锁，但存在局限性：

[FunctionName("OperationalFunction")]
public async Task OperationalFunction(
    [ActivityTrigger] string input,
    ILogger log,
    [Redis("your-redis-connection-string")] IDistributedCache cache)
{
    var lockKey = "OperationalFunctionLock";
    var lockOptions = new DistributedCacheEntryOptions { AbsoluteExpirationRelativeToNow = TimeSpan.FromMinutes(10) };

    // 轮询获取锁
    while (await cache.GetStringAsync(lockKey) != null)
    {
        await Task.Delay(500);
    }

    try
    {
        await cache.SetStringAsync(lockKey, "locked", lockOptions);
        // 你的业务逻辑代码
        log.LogInformation($"Executing task: {input}");
    }
    finally
    {
        await cache.RemoveAsync(lockKey);
    }
}

局限性： #

• 需要额外依赖Azure Redis Cache，增加成本和复杂度
• 无法严格保证FIFO顺序（不同请求的重试时机可能打乱顺序）
• 函数崩溃时需依赖锁过期时间释放，存在短暂死锁风险

原方案无效原因说明 #

• host.json的blobs.maxDegreeOfParallelism仅控制BlobTrigger自身的并行度，无法限制HttpTrigger及后续编排/Activity的并行
• 轮询编排状态的方式效率极低，且无法避免多请求同时检测到编排完成、同时启动新实例的情况，无法保证严格串行

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