Azure中如何基于Service Bus消息的Key触发不同Azure Function？代码验证

当前代码的核心问题 #

• 你给三个函数都添加了@app.service_bus_queue_trigger装饰器，且指向同一个队列queue-asb，这会导致每条消息发送到队列时，三个函数都会被Azure Functions runtime同时触发执行，完全不符合你"根据key触发不同函数"的需求。
• 你在主函数里直接调用service_bus_queue()和servicebus_trigger()这两个带触发器的函数，这是错误的——带触发器装饰器的函数由Azure runtime管理，不能像普通Python函数那样直接调用。

方案一：单触发器+内部分支（适合简单场景） #

保留一个队列触发器作为入口，把另外两个函数改成普通处理函数（去掉触发器装饰器），根据消息的key值分支调用对应逻辑：

import azure.functions as func
import logging
import json

app = func.FunctionApp()

# 仅保留一个队列触发器作为消息入口
@app.service_bus_queue_trigger(arg_name="azservicebus", queue_name="queue-asb",
                               connection="azuretopicnamespace_SERVICEBUS") 
def servicebus_queue_trigger(azservicebus: func.ServiceBusMessage):
    try:
        message_body = azservicebus.get_body().decode('utf-8')
        message_data = json.loads(message_body)
        key_val = message_data.get('key_val', '')
        
        if key_val == 'service_bus_queue':
            logging.info("触发service_bus_queue处理逻辑")
            result = handle_service_bus_queue(azservicebus)
            logging.info(f"处理结果: {result}")
        else:
            logging.info("触发servicebus_trigger处理逻辑")
            handle_servicebus_trigger(azservicebus)
    except Exception as e:
        logging.error(f"消息处理失败: {str(e)}")

# 普通处理函数，不带触发器装饰器
def handle_service_bus_queue(azservicebus: func.ServiceBusMessage):
    a = 1 + 2
    logging.info(a)
    logging.info("这是第一个处理逻辑")
    return a

# 普通处理函数，不带触发器装饰器
def handle_servicebus_trigger(azservicebus: func.ServiceBusMessage):
    logging.info("这是第二个处理逻辑")

方案二：使用Service Bus主题+订阅（推荐，适合复杂/扩展场景） #

如果后续业务需要更多分支逻辑，更符合Azure事件驱动架构的方式是用主题+订阅：

1. 创建一个Service Bus主题（比如my-topic）
2. 为每个需要触发的逻辑创建独立订阅：
   • 订阅1：设置过滤规则key_val = 'service_bus_queue'
   • 订阅2：设置过滤规则key_val != 'service_bus_queue'（或其他自定义规则）
3. 每个Function绑定对应的订阅触发器，Azure runtime会自动根据过滤规则把消息路由到对应Function执行

示例代码（两个独立的Function，分别绑定不同订阅）：

import azure.functions as func
import logging
import json

app = func.FunctionApp()

# 绑定到订阅1，处理key_val=service_bus_queue的消息
@app.service_bus_topic_trigger(arg_name="azservicebus", topic_name="my-topic",
                               subscription_name="sub-service-bus-queue",
                               connection="azuretopicnamespace_SERVICEBUS") 
def handle_service_bus_queue_topic(azservicebus: func.ServiceBusMessage):
    a = 1 + 2
    logging.info(a)
    logging.info("这是第一个订阅的处理逻辑")

# 绑定到订阅2，处理其他消息
@app.service_bus_topic_trigger(arg_name="azservicebus", topic_name="my-topic",
                               subscription_name="sub-other",
                               connection="azuretopicnamespace_SERVICEBUS") 
def handle_other_topic(azservicebus: func.ServiceBusMessage):
    logging.info("这是第二个订阅的处理逻辑")

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