使用Python 3.x无法连接Azure ServiceBus问题排查

咱们先从最容易排查的点入手，连接类问题大多和配置、依赖或权限有关：

一、确认SDK版本与代码匹配 #

Azure ServiceBus的Python SDK有两个差异极大的版本：**v7+（最新异步优先版本）**和旧版v2。如果你按最新文档安装了v7+，但代码是旧版写法，必然会报错：

• 先检查当前安装的版本：

  pip show azure-servicebus
  

• 如果是v7+，请使用对应API编写代码，示例如下：

  from azure.servicebus import ServiceBusClient, ServiceBusMessage
  
  CONNECTION_STR = "你的命名空间连接字符串"
  QUEUE_NAME = "目标队列名称"
  
  def send_test_message():
      with ServiceBusClient.from_connection_string(conn_str=CONNECTION_STR) as client:
          with client.get_queue_sender(queue_name=QUEUE_NAME) as sender:
              message = ServiceBusMessage("Hello ServiceBus!")
              sender.send_messages(message)
              print("消息发送成功")
  

• 如果你的代码基于旧版v2，就指定安装对应版本：

  pip install azure-servicebus==2.1.1
  

二、验证核心配置正确性 #

这是最容易踩坑的环节：

• 确保连接字符串是从Azure门户的Service Bus命名空间获取的（不是队列/主题本身的），格式类似：
  Endpoint=sb://<命名空间>.servicebus.windows.net/;SharedAccessKeyName=<密钥名称>;SharedAccessKey=<密钥值>
• 检查队列/主题名称完全匹配，注意大小写敏感（Azure ServiceBus实体名称区分大小写）
• 确认使用的Shared Access Key拥有对应权限（比如发送/接收权限，避免因权限不足报错）

三、网络与环境排查 #

如果运行环境在企业内网或有防火墙限制：

• 确保出站端口5671（AMQP协议）和443（HTTPS协议）处于开放状态，Azure ServiceBus依赖这些端口
• 如果在代理环境下，需要在代码中配置代理：

  from azure.servicebus import ServiceBusClient
  from azure.core.credentials import AzureNamedKeyCredential
  import os
  
  credential = AzureNamedKeyCredential(os.environ["SB_SAS_NAME"], os.environ["SB_SAS_KEY"])
  client = ServiceBusClient(
      fully_qualified_namespace=os.environ["SB_FQDN"],
      credential=credential,
      proxy_config={
          "proxy_url": "http://你的代理地址:端口",
          "username": "代理用户名",
          "password": "代理密码"
      }
  )
  

四、捕获具体错误信息 #

如果上述步骤都没解决问题，建议添加异常捕获拿到精准错误内容，方便定位：

from azure.servicebus import ServiceBusClient, ServiceBusMessage
from azure.servicebus.exceptions import ServiceBusError

CONNECTION_STR = "你的连接字符串"
QUEUE_NAME = "目标队列名称"

try:
    with ServiceBusClient.from_connection_string(conn_str=CONNECTION_STR) as client:
        with client.get_queue_sender(queue_name=QUEUE_NAME) as sender:
            message = ServiceBusMessage("测试消息")
            sender.send_messages(message)
            print("消息发送成功")
except ServiceBusError as e:
    print(f"ServiceBus服务端错误: {str(e)}")
except Exception as e:
    print(f"意外错误: {str(e)}")

把捕获到的错误信息贴出来，就能更针对性地解决问题啦。

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