如何通过代码对Azure Function中高负载的Service Bus函数单独限流？

可以通过代码对特定高负载的Service Bus触发Azure Functions单独限流，以下是几种实用的实现方案：

方案一：本地信号量（SemaphoreSlim）单实例限流 #

适合单实例部署或对跨实例限流要求不高的场景，通过信号量控制单个函数的并发执行数。

// 针对当前函数的限流信号量，限制最多10个并发任务
private static readonly SemaphoreSlim _functionSemaphore = new SemaphoreSlim(10);

[FunctionName("HighLoadServiceBusFunc")]
public async Task Run(
    [ServiceBusTrigger("high-load-queue", Connection = "ServiceBusConn")] string queueItem,
    ILogger log)
{
    await _functionSemaphore.WaitAsync();
    try
    {
        // 业务处理逻辑
        log.LogInformation($"Processing message: {queueItem}");
        await Task.Delay(1000); // 模拟耗时操作
    }
    finally
    {
        _functionSemaphore.Release();
    }
}

• 注意：此方案仅在单个函数实例内生效，多实例部署时每个实例会独立控制并发数，若需全局限流需结合分布式方案。

方案二：分布式Redis全局限流 #

针对多实例部署场景，利用Redis实现跨实例的统一并发控制，避免单个实例限流导致的全局资源耗尽。

private static IDatabase _redisDatabase;

static HighLoadServiceBusFunc()
{
    var redisConn = ConnectionMultiplexer.Connect(Environment.GetEnvironmentVariable("RedisConnString"));
    _redisDatabase = redisConn.GetDatabase();
}

[FunctionName("HighLoadServiceBusFunc")]
public async Task Run(
    [ServiceBusTrigger("high-load-queue", Connection = "ServiceBusConn")] string queueItem,
    ILogger log)
{
    string limitKey = "func:high-load:concurrent-limit";
    int maxGlobalConcurrent = 50; // 全局最大并发数

    var currentCount = await _redisDatabase.StringIncrementAsync(limitKey);
    try
    {
        if (currentCount > maxGlobalConcurrent)
        {
            log.LogWarning($"Global rate limit hit, message skipped: {queueItem}");
            // 抛出异常让Service Bus重新排队（根据业务需求选择是否重试）
            throw new InvalidOperationException("Rate limit exceeded");
        }

        // 业务处理逻辑
        log.LogInformation($"Processing message: {queueItem}");
        await Task.Delay(1000);
    }
    finally
    {
        // 递减计数器，避免异常导致计数残留
        await _redisDatabase.StringDecrementAsync(limitKey);
        // 可选：设置键过期时间，防止异常场景下计数无法重置
        await _redisDatabase.KeyExpireAsync(limitKey, TimeSpan.FromMinutes(5));
    }
}

• 注意：需处理Redis连接异常，确保计数能正确递减，避免出现“僵尸计数”。

方案三：批量消息处理+并发控制 #

结合Service Bus的批量接收特性，在代码中控制批量处理的并发数，同时配合host.json的函数级配置优化。

代码实现 #

[FunctionName("HighLoadServiceBusFunc")]
public async Task Run(
    [ServiceBusTrigger("high-load-queue", Connection = "ServiceBusConn", 
        AutoCompleteMessages = false, MaxAutoRenewDuration = "00:05:00")] 
    ServiceBusReceivedMessage[] batchMessages,
    ServiceBusMessageActions messageActions,
    ILogger log)
{
    var batchSemaphore = new SemaphoreSlim(20); // 单次批量处理最多20条并发
    var processTasks = new List<Task>();

    foreach (var msg in batchMessages)
    {
        await batchSemaphore.WaitAsync();
        processTasks.Add(Task.Run(async () =>
        {
            try
            {
                // 单条消息处理逻辑
                log.LogInformation($"Processing message ID: {msg.MessageId}");
                await Task.Delay(500);
                await messageActions.CompleteMessageAsync(msg);
            }
            catch (Exception ex)
            {
                log.LogError(ex, $"Failed to process message {msg.MessageId}");
                await messageActions.AbandonMessageAsync(msg);
            }
            finally
            {
                batchSemaphore.Release();
            }
        }));
    }

    await Task.WhenAll(processTasks);
}

配合host.json函数级配置 #

{
  "version": "2.0",
  "extensions": {
    "serviceBus": {
      "functions": {
        "HighLoadServiceBusFunc": {
          "maxConcurrentCalls": 50,
          "prefetchCount": 100
        }
      }
    }
  }
}

• 此方案通过批量接收减少连接开销，同时代码层控制并发处理数，平衡吞吐量与资源占用。

方案四：Polly速率限制策略 #

利用Polly库的RateLimit策略，对函数的执行频率进行限制，适合控制每秒/每分钟的处理次数。

private static readonly RateLimitPolicy _rateLimitPolicy = Policy.RateLimitAsync(
    100, // 每分钟最多处理100次
    TimeSpan.FromMinutes(1));

[FunctionName("HighLoadServiceBusFunc")]
public async Task Run(
    [ServiceBusTrigger("high-load-queue", Connection = "ServiceBusConn")] string queueItem,
    ILogger log)
{
    try
    {
        await _rateLimitPolicy.ExecuteAsync(async () =>
        {
            // 业务处理逻辑
            log.LogInformation($"Processing message: {queueItem}");
            await Task.Delay(1000);
        });
    }
    catch (RateLimitRejectedException)
    {
        log.LogWarning($"Rate limit exceeded, message will be retried");
        // 抛出异常触发Service Bus重试机制
        throw;
    }
}

• 注意：Polly的本地速率限制仅在单实例生效，多实例场景需结合分布式限流方案使用。

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