如何限制服务总线队列触发的Azure Function调用频率适配第三方API限制？

这个问题我之前帮好几个开发者解决过，核心就是要把Azure Function调用第三方API的频率给稳稳压在每秒10次以内，下面给你几个从简单到进阶的实用方案，你可以根据自己的场景选：

方案1：调整Azure Function宿主的并发设置（最简单直接） #

服务总线触发的Function默认会同时处理多条消息，你可以通过host.json的配置直接限制并发处理的消息数量——如果你的Function处理单条消息+调用API的耗时接近1秒，把并发数设为10刚好能满足每秒10次的限制。

示例host.json配置：

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

• maxConcurrentCalls：设置为10，意味着同一时间最多有10个Function实例在处理消息
• prefetchCount：设置成和maxConcurrentCalls一致，避免服务总线给Function推送过多消息导致积压

⚠️ 注意：如果你的Function执行时间远短于1秒（比如200ms），那这个设置会导致每秒产生50次API调用，超过限制。这种情况就得结合其他方案来精准控制速率。

方案2：在代码层加入速率限制逻辑（更精准的控制） #

如果宿主并发设置不够灵活，或者Function执行时间不稳定，那可以在代码里结合并发控制和时间窗口限流来严格卡每秒10次的限制。以C#为例，用SemaphoreSlim控制并发请求数，再配合时间窗口计数器：

public static class QueueProcessingFunction
{
    // 静态信号量：限制最多10个并发API请求
    private static readonly SemaphoreSlim _apiConcurrencySemaphore = new SemaphoreSlim(10, 10);
    // 时间窗口限流相关变量
    private static int _currentSecondRequests = 0;
    private static DateTime _windowStartTime = DateTime.UtcNow;
    private static readonly object _lockObject = new object();

    [FunctionName("ProcessServiceBusMessage")]
    public static async Task Run(
        [ServiceBusTrigger("your-queue-name", Connection = "ServiceBusConnectionString")] string queueMessage,
        ILogger log)
    {
        // 先处理时间窗口限流：确保每秒不超过10次请求
        lock (_lockObject)
        {
            var now = DateTime.UtcNow;
            // 如果当前时间超出了1秒窗口，重置计数和窗口时间
            if (now - _windowStartTime >= TimeSpan.FromSeconds(1))
            {
                _windowStartTime = now;
                _currentSecondRequests = 0;
            }

            // 如果当前窗口已经到10次，等待到下一秒再继续
            if (_currentSecondRequests >= 10)
            {
                var waitTime = _windowStartTime.AddSeconds(1) - now;
                Thread.Sleep(waitTime);
                // 重置窗口
                _windowStartTime = DateTime.UtcNow;
                _currentSecondRequests = 0;
            }
            _currentSecondRequests++;
        }

        // 等待信号量，确保并发请求不超10个
        await _apiConcurrencySemaphore.WaitAsync();
        try
        {
            // 调用第三方API的逻辑
            await CallThirdPartyApi(queueMessage);
        }
        finally
        {
            // 释放信号量，让后续请求可以进入
            _apiConcurrencySemaphore.Release();
        }
    }

    private static async Task CallThirdPartyApi(string messageData)
    {
        // 这里替换成你的第三方API调用代码
        using var httpClient = new HttpClient();
        await httpClient.PostAsync("https://third-party-api.com/endpoint", new StringContent(messageData));
    }
}

其他语言也有对应的工具，比如Python可以用ratelimit库，Node.js可以用bottleneck库，原理都是一样的：控制时间窗口内的请求数+并发数。

方案3：用Durable Functions实现精确速率控制（适合高负载复杂场景） #

如果你的服务总线队列消息量很大，或者需要更灵活的速率调整，Durable Functions是个绝佳选择。你可以通过Orchestrator函数严格控制Activity函数的执行速率，确保每秒只调用10次第三方API：

大致流程： #

1. 用服务总线触发的Function接收消息，将消息转发给Durable Orchestrator
2. Orchestrator函数按批次处理消息：每次处理10条，然后等待1秒，再处理下一批

示例代码（C#）：

Orchestrator函数

[FunctionName("RateLimitedProcessingOrchestrator")]
public static async Task RunOrchestrator(
    [OrchestrationTrigger] IDurableOrchestrationContext context)
{
    // 接收从触发器传来的消息列表
    var messages = context.GetInput<List<string>>();
    const int batchSize = 10;
    // 将消息分成每10条一组
    var messageBatches = messages.Chunk(batchSize);

    foreach (var batch in messageBatches)
    {
        // 并行处理当前批次的10条消息（调用第三方API）
        var apiCallTasks = batch.Select(msg => context.CallActivityAsync("InvokeThirdPartyApi", msg));
        await Task.WhenAll(apiCallTasks);
        // 等待1秒，再处理下一批
        await context.CreateTimer(context.CurrentUtcDateTime.AddSeconds(1), CancellationToken.None);
    }
}

服务总线触发的Function

[FunctionName("ServiceBusToDurableTrigger")]
public static async Task Run(
    [ServiceBusTrigger("your-queue-name", Connection = "ServiceBusConnectionString")] string queueMessage,
    [DurableClient] IDurableOrchestrationClient durableClient,
    ILogger log)
{
    // 这里可以优化成批量收集消息后再启动Orchestrator，避免每次单条启动
    var instanceId = await durableClient.StartNewAsync("RateLimitedProcessingOrchestrator", new List<string> { queueMessage });
    log.LogInformation($"Started rate-limited orchestration with ID: {instanceId}");
}

Activity函数（实际调用API的逻辑）

[FunctionName("InvokeThirdPartyApi")]
public static async Task CallApi([ActivityTrigger] string message, ILogger log)
{
    // 第三方API调用逻辑
    using var httpClient = new HttpClient();
    var response = await httpClient.PostAsync("https://third-party-api.com/endpoint", new StringContent(message));
    response.EnsureSuccessStatusCode();
}

这个方案的好处是Durable Functions会自动处理重试、持久化，即使Function重启也不会丢失消息，非常适合高负载场景。

方案4：通过Azure API Management做限流（如果允许代理API） #

如果你的第三方API允许通过Azure API Management（APIM）代理调用，那可以直接在APIM里配置速率限制策略，让APIM帮你把请求压到每秒10次。这种方式不需要修改Function代码，只需要把Function里的API调用地址改成APIM的网关地址即可。

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