同一Azure Function App多函数运行模式及串行执行实现咨询

是否可行 #

完全可行。默认情况下，采用服务计划的Azure Function App中，两个Service Bus触发的函数会并行执行（只要有消息待处理）。要实现交替顺序处理，需要通过并发限制和同步锁机制来约束函数的执行时机。

具体实现步骤 #

1. 限制Function App的并发基础环境 #

首先确保整个Function App在单进程、单实例下运行，避免多实例/多进程带来的并行执行：

• 在应用设置中添加以下配置：
  • FUNCTIONS_WORKER_PROCESS_COUNT = 1：限制每个实例的工作进程数为1
  • WEBSITE_MAX_DYNAMIC_APPLICATION_SCALE_OUT = 1：禁止自动扩缩容，固定为1个实例

2. 配置Service Bus触发器的单消息处理 #

修改host.json，让每个Service Bus触发的函数一次只处理一条消息，避免预取多条导致的并行：

{
  "version": "2.0",
  "extensions": {
    "serviceBus": {
      "prefetchCount": 1,
      "maxConcurrentCalls": 1
    }
  }
}

• maxConcurrentCalls = 1：每个函数同时处理的消息数限制为1
• prefetchCount = 1：仅预取1条消息，避免提前获取多条消息导致后续并行处理

3. 添加全局同步锁 #

通过锁机制确保两个函数同一时间只有一个在执行，实现交替处理。根据部署场景选择本地锁或分布式锁：

场景一：单实例部署（推荐，符合你的顺序处理需求）

使用静态SemaphoreSlim作为本地锁（支持异步操作），两个函数共享同一个锁对象：

Function1代码示例：

using System.Threading;
using System.Threading.Tasks;
using Microsoft.Azure.WebJobs;
using Microsoft.Extensions.Logging;

namespace YourFunctionApp
{
    public static class Function1
    {
        // 全局共享的信号量，初始许可数为1，最大许可数为1
        private static readonly SemaphoreSlim _processingLock = new SemaphoreSlim(1, 1);

        [FunctionName("Function1")]
        public static async Task Run(
            [ServiceBusTrigger("queue1", Connection = "ServiceBusConnection")] string message,
            ILogger log)
        {
            await _processingLock.WaitAsync();
            try
            {
                // 处理queue1消息的业务逻辑
                log.LogInformation($"处理queue1消息：{message}");
                // 模拟异步处理耗时
                await Task.Delay(1000);
            }
            finally
            {
                // 释放锁，允许另一个函数执行
                _processingLock.Release();
            }
        }
    }
}

Function2代码示例（仅触发器队列名不同，锁对象与Function1完全一致）：

using System.Threading;
using System.Threading.Tasks;
using Microsoft.Azure.WebJobs;
using Microsoft.Extensions.Logging;

namespace YourFunctionApp
{
    public static class Function2
    {
        // 必须与Function1使用同一个静态锁对象
        private static readonly SemaphoreSlim _processingLock = new SemaphoreSlim(1, 1);

        [FunctionName("Function2")]
        public static async Task Run(
            [ServiceBusTrigger("queue2", Connection = "ServiceBusConnection")] string message,
            ILogger log)
        {
            await _processingLock.WaitAsync();
            try
            {
                // 处理queue2消息的业务逻辑
                log.LogInformation($"处理queue2消息：{message}");
                // 模拟异步处理耗时
                await Task.Delay(1000);
            }
            finally
            {
                _processingLock.Release();
            }
        }
    }
}

场景二：多实例部署（若未来需要扩实例）

如果需要支持多实例，本地锁会失效，需使用分布式锁，比如Azure Blob租约：

using Azure.Storage.Blobs;
using Azure.Storage.Blobs.Specialized;
using System.Threading.Tasks;
using Microsoft.Azure.WebJobs;
using Microsoft.Extensions.Logging;

namespace YourFunctionApp
{
    public static class Function1
    {
        private static readonly BlobContainerClient _containerClient = new BlobContainerClient(
            Environment.GetEnvironmentVariable("BlobStorageConnection"), 
            "processing-locks");
        private static readonly string _lockBlobName = "global-processing-lock";

        [FunctionName("Function1")]
        public static async Task Run(
            [ServiceBusTrigger("queue1", Connection = "ServiceBusConnection")] string message,
            ILogger log)
        {
            // 确保容器存在
            await _containerClient.CreateIfNotExistsAsync();
            BlobClient lockBlob = _containerClient.GetBlobClient(_lockBlobName);
            await lockBlob.CreateIfNotExistsAsync();

            BlobLeaseClient leaseClient = lockBlob.GetBlobLeaseClient();
            try
            {
                // 获取5分钟的租约（根据你的消息处理耗时调整）
                await leaseClient.AcquireAsync(TimeSpan.FromMinutes(5));
                // 处理queue1消息的业务逻辑
                log.LogInformation($"处理queue1消息：{message}");
                await Task.Delay(1000);
            }
            finally
            {
                // 释放租约
                await leaseClient.ReleaseAsync();
            }
        }
    }
}

Function2使用完全相同的Blob租约逻辑即可，确保同一时间只有一个实例的函数能拿到租约执行。

效果验证 #

当两个队列各有200条消息时，两个函数会交替获取锁处理消息（先处理queue1的一条，再处理queue2的一条，依此类推），不会出现并行执行的情况，完全符合你“如同单个函数依次从两个队列取消息”的需求。

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