Azure Function循环处理Service Bus消息时无法向Topic发送问题求助

• 重复消费逻辑冲突：你已经使用了Azure Function Service Bus触发器监听TOPIC_A，触发器会自动消费主题中的消息并通过main函数的message参数传入。你代码中额外创建了订阅接收器主动拉取TOPIC_A的消息，部署到Azure后触发器会先消费消息，导致你自定义的接收器拉不到任何消息，received_msgs为空数组，for msg in received_msgs循环直接跳过，内部的发送逻辑自然不会执行。本地测试时没有触发器自动消费逻辑，自定义接收器能拉到消息，所以运行正常。
• 连接实例化逻辑错误：你在自定义接收器的with块内部，又对servicebus_client变量重复赋值、新建连接，会覆盖外层的接收器连接，直接触发接收器提前关闭，和你日志中看到的接收器状态变更、连接关闭信息完全对应。
• 不合理的函数定义：send_output函数定义在for循环内部，每次循环都会重复创建函数实例，且循环内重复创建ServiceBus客户端连接会产生不必要的性能开销，也容易引发作用域问题。

直接使用触发器传入的消息即可，无需额外自定义接收器拉取消息，简化后的核心逻辑如下：

import logging
import azure.functions as func
import json
import boto3
from azure.keyvault.secrets import SecretClient
from azure.identity import DefaultAzureCredential
from azure.servicebus import ServiceBusClient, ServiceBusMessage

# 把可复用的初始化逻辑提到函数外，避免每次触发都重复初始化
KeyVault_Url = '你的KeyVault地址'
credential = DefaultAzureCredential()
client_keyvault = SecretClient(vault_url=KeyVault_Url, credential=credential)
CONNECTION_STR = client_keyvault.get_secret("CONN").value
comprehend = boto3.client(
    service_name='comprehend', 
    region_name='eu-west-1', 
    aws_access_key_id=client_keyvault.get_secret("ID").value, 
    aws_secret_access_key=client_keyvault.get_secret("SECRET").value
)
TOPIC_NAME_B = "TOPICB" 

def main(message: func.ServiceBusMessage):
    logging.info("接收到TOPIC_A消息")
    # 直接处理触发器传入的消息，不需要自己拉取
    message_str = message.get_body().decode('utf-8')
    res = json.loads(message_str) 
    text = res['Text']

    # 调用comprehend处理逻辑和原来一致
    result = comprehend.detect_sentiment(Text=text, LanguageCode='en')
    sentiment = result["Sentiment"]
    if sentiment == "POSITIVE":
        value = round(result["SentimentScore"]["Positive"] * 100,2)
    elif sentiment == "NEGATIVE":
        value = round(result["SentimentScore"]["Negative"] * 100,2)
    elif sentiment == "NEUTRAL":
        value = round(result["SentimentScore"]["Neutral"] * 100,2)
    elif sentiment == "MIXED":
        value = round(result["SentimentScore"]["Mixed"] * 100,2)

    lang_result_json = comprehend.detect_dominant_language(Text = text)
    for line in lang_result_json["Languages"]:
        line['Score'] = round(line['Score']* 100, 2)
    
    output = {
        'XXX': res['XXX'],
        'XXX Id': res['XXX'],
        'XXX': res['XXX'],
        'XXX': res['XXX'],
        'XXX': res['XXX'],
        'Sentiment': sentiment,
        'Value': value,
        'Languages': lang_result_json['Languages']
    }
    output_json = json.dumps(output, ensure_ascii=False)

    # 发送到TOPIC_B
    with ServiceBusClient.from_connection_string(conn_str=CONNECTION_STR, logging_enable=True) as servicebus_client:
        sender = servicebus_client.get_topic_sender(topic_name=TOPIC_NAME_B)
        with sender:
            message2 = ServiceBusMessage(
                output_json,
                content_type="XXX",
                ApplicationProperties={b'tenantcode':message.application_properties[b'tenantcode']}
            )
            sender.send_messages(message2)
    logging.info("消息处理完成，已转发到TOPIC_B")

• 可将ServiceBus客户端的初始化也提到函数外，复用连接减少性能开销
• 如果需要批量处理消息，可修改触发器配置为批量接收模式，无需自己实现拉取逻辑

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