Python版Azure Service Bus队列触发器无法触发及本地检测异常

一、本地"No job functions found"错误根源排查 #

本地无法识别函数是部署后触发器失效的核心原因，按以下顺序排查：

1. 核对文件结构与函数装饰器规范 #

• 确保function_app.py位于项目根目录，而非子文件夹中
• 检查函数装饰器的正确性：
  • HTTP函数需使用@app.route(route="get-all-data", methods=["GET"])（v2模型），装饰器拼写、路由名称不能出错
  • Service Bus队列触发器需使用@app.service_bus_queue_trigger(arg_name="msg", queue_name="你的队列名", connection="ServiceBusConnection")，参数queue_name要与Azure中创建的队列名完全一致，connection要匹配配置文件中的连接字符串键名

2. 验证local.settings.json配置 #

必须包含以下关键配置，尤其是v2模型强制要求的索引开关：

{
  "IsEncrypted": false,
  "Values": {
    "AzureWebJobsStorage": "UseDevelopmentStorage=true",
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "AzureWebJobsFeatureFlags": "EnableWorkerIndexing",
    "ServiceBusConnection": "你的Service Bus连接字符串"
  }
}

• 确认FUNCTIONS_WORKER_RUNTIME值为python，不能填其他语言
• 本地运行时，若使用Azure存储模拟器，需确保模拟器处于启动状态

3. 检查依赖与Python版本 #

• 确保requirements.txt包含必要依赖包：

azure-functions
azure-servicebus
# 数据库相关依赖，比如用SQL则添加azure-sql-management或pyodbc

• 本地Python版本需与Azure函数应用运行时版本一致（推荐3.8-3.11，可在Azure门户函数应用的"配置"中查看）
• 执行pip install -r requirements.txt，确保依赖安装在当前使用的虚拟环境中

4. 核对host.json配置 #

v2模型基础配置如下，重点确认扩展包版本支持Service Bus触发器：

{
  "version": "2.0",
  "logging": {
    "applicationInsights": {
      "samplingSettings": {
        "isEnabled": true,
        "excludedTypes": "Request"
      }
    }
  },
  "extensionBundle": {
    "id": "Microsoft.Azure.Functions.ExtensionBundle",
    "version": "[3.*, 4.0.0)"
  }
}

二、部署后Service Bus触发器失效补充排查 #

1. 登录Azure门户，检查函数应用的"配置"→"应用设置"中，ServiceBusConnection的值是否为拥有队列读写权限的连接字符串
2. 查看函数应用日志：进入"监控"→"日志"，搜索是否有连接失败、权限不足的报错信息
3. 确认Service Bus队列的活跃消息数不为0，消息确实已入队
4. 在Azure门户的函数详情页，检查触发器配置中的队列名称、连接字符串键名是否与代码一致

三、正确代码示例参考 #

function_app.py #

import azure.functions as func
import logging

app = func.FunctionApp()

# HTTP路由函数：get-all-data
@app.route(route="get-all-data", methods=["GET"])
def get_all_data(req: func.HttpRequest) -> func.HttpResponse:
    logging.info('Python HTTP trigger processed request.')
    # 插入数据库全量查询逻辑
    return func.HttpResponse("全量数据已获取", status_code=200)

# Service Bus队列触发器
@app.service_bus_queue_trigger(arg_name="msg", queue_name="your-queue-name", connection="ServiceBusConnection")
def sb_queue_trigger(msg: func.ServiceBusMessage):
    logging.info('Processed Service Bus message: %s', msg.get_body().decode('utf-8'))
    # 插入数据库存储消息逻辑

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