如何配置Azure Function App并行处理Service Bus队列消息？

我已创建Azure Function App，并通过Azure容器注册表中的Docker镜像创建了函数，该函数由Azure Service Bus队列的新消息触发。当前函数每次执行（单条队列消息触发）耗时较长，有时可达1小时，但队列消息量大时函数无法自动缩放：一次性放入100条消息，总执行时长竟达约100小时（串行处理），且函数实例远未达到允许的最大并发数。

当前配置文件 #

host.json #

{
  "version": "2.0",
  "logging": {
    "applicationInsights": {
      "samplingSettings": {
        "isEnabled": true,
        "excludedTypes": "Request"
      }
    }
  },
  "functionTimeout": "05:00:00",
  "extensionBundle": {
    "id": "Microsoft.Azure.Functions.ExtensionBundle",
    "version": "[3.*, 4.0.0)"
  },
  "extensions": {
      "serviceBus": {
      "prefetchCount": 1,
      "messageHandlerOptions": {
          "autoComplete": true,
          "maxConcurrentCalls": 1,
          "maxAutoRenewDuration": "00:05:00"
      },
      "sessionHandlerOptions": {
          "autoComplete": false,
          "messageWaitTimeout": "00:00:30",
          "maxAutoRenewDuration": "00:55:00",
          "maxConcurrentSessions": 16
      },
      "batchOptions": {
          "maxMessageCount": 1000,
          "operationTimeout": "00:01:00",
          "autoComplete": true
      }
      }
  }    
}

function.json #

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "name": "msgIn",
      "type": "serviceBusTrigger",
      "isSessionsEnabled": true,
      "direction": "in",
      "queueName": "queuename",
      "connection": "AzureServiceBusConnectionString",
    }
  ]
}

函数代码 #

def main(msgIn: func.ServiceBusMessage):

    payload = msgIn.get_body().decode('utf-8')
    payload_json = json.loads(payload)
    # do things to payload_json
    # ...
    return

注：队列queuename已启用会话功能。

请问当前配置是否符合我的使用场景？若不符合，应如何修改？我已尝试多种配置调整但均无效，Azure相关配置的语义文档不够清晰。

当前配置不符合你的使用场景，核心问题出在会话队列的并发控制逻辑以及Python函数的单实例并发限制上，具体调整如下：

1. 核心问题定位 #

• 启用会话的队列会忽略messageHandlerOptions配置，所有并发控制由sessionHandlerOptions决定，但当前Python函数默认单实例单进程，即使maxConcurrentSessions设为16，单实例也只能串行处理会话。
• maxAutoRenewDuration设为55分钟，小于你的函数最长执行时长（1小时），会导致会话锁提前过期，引发重复处理。
• prefetchCount=1会限制单会话内的预取消息数，增加网络往返等待时间。

2. 具体修改步骤 #

修改host.json配置 #

{
  "version": "2.0",
  "logging": {
    "applicationInsights": {
      "samplingSettings": {
        "isEnabled": true,
        "excludedTypes": "Request"
      }
    }
  },
  "functionTimeout": "05:00:00",
  "functionAppScaleLimit": 20, // 允许的最大实例数，按需调整
  "workers": {
    "python": {
      "workerCount": "auto", // 自动根据CPU核心数调整工作进程数
      "maxWorkerCount": 8 // 单实例最大工作进程数，按需设置
    }
  },
  "extensionBundle": {
    "id": "Microsoft.Azure.Functions.ExtensionBundle",
    "version": "[3.*, 4.0.0)"
  },
  "extensions": {
      "serviceBus": {
      "prefetchCount": 10, // 提高预取数，减少会话内消息拉取等待
      "sessionHandlerOptions": {
          "autoComplete": true, // 函数执行完成后自动标记消息完成
          "messageWaitTimeout": "00:01:00", // 延长会话等待超时，避免频繁断开
          "maxAutoRenewDuration": "01:05:00", // 延长会话锁续期时间，覆盖最长执行时长
          "maxConcurrentSessions": 32 // 提高单实例并发会话数，按需调整
      }
      }
  }    
}

• workers.python配置：解决Python函数单实例单进程的限制，auto模式会根据实例CPU核心数自动分配工作进程，让单实例能同时处理多个会话。
• sessionHandlerOptions调整：
  • autoComplete改为true，避免手动处理消息完成的遗漏风险。
  • maxAutoRenewDuration设为1小时5分钟，确保函数执行期间会话锁不会过期。
  • maxConcurrentSessions调高到32，提升单实例的会话并发处理能力。
• prefetchCount：提高到10，减少会话内频繁拉取消息的网络开销。

确认function.json配置 #

保持isSessionsEnabled: true不变，确保绑定到会话队列的配置正确。

3. 额外优化建议 #

• 实例缩放配置：在Azure门户的Function App「缩放」设置中，将「最大实例数」调高到合适值（比如20），允许平台根据队列消息量自动扩容实例。
• 会话ID均匀分配：如果业务允许，将消息均匀分配到不同会话ID中，避免单会话消息堆积（会话内的消息必须串行处理）。
• 异步函数改造：如果业务逻辑允许，将函数改为异步模式，进一步提升单进程内的并发处理能力。

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