使用Python脚本获取Azure Service Bus最后一条消息属性时遇会话错误

错误原因 #

你的Service Bus订阅SubCardDBUpdater_20启用了**会话（Session）**机制，而当前脚本创建的是普通非会话消息接收器，违反了Service Bus的规则——启用会话的实体必须使用会话接收器访问消息，因此抛出ErrorCodes.NotAllowed错误。

修复后的脚本 #

针对会话订阅，需修改接收器创建方式，使用会话接收器接收消息。以下是两种场景的修复方案：

场景1：接收任意可用会话的最后一条消息 #

无需指定特定会话ID时，用accept_next_session()获取下一个可用的会话接收器：

#!/usr/bin/python3

from azure.servicebus import ServiceBusClient
from azure.identity import DefaultAzureCredential

def get_last_message_properties(namespace_name, topic_name, subscription_name):
    try:
        # 创建Service Bus客户端
        servicebus_client = ServiceBusClient(
            fully_qualified_namespace=namespace_name,
            credential=DefaultAzureCredential()
        )

        # 获取订阅的会话接收器（自动接受下一个可用会话）
        with servicebus_client:
            with servicebus_client.accept_next_session(
                topic_name=topic_name,
                subscription_name=subscription_name
            ) as session_receiver:
                # 接收会话中的最后1条消息
                messages = session_receiver.receive(max_message_count=1)
                for message in messages:
                    try:
                        # 访问消息属性
                        properties = message.properties
                        cdId = properties.get("cdId")
                        active = properties.get("active")
                        alarm = properties.get("alarm")
                        print(f"会话 {session_receiver.session_id} 的最后消息属性: cdId={cdId}, active={active}, alarm={alarm}")
                        # 完成消息（若不需要保留消息）
                        message.complete()
                    except Exception as e:
                        print(f"处理消息时出错: {e}")

    except Exception as e:
        print(f"整体执行出错: {e}")

if __name__ == "__main__":
    namespace = "sbdev.servicebus.windows.net"
    topic = "topicsystemlog_main"
    subscription = "SubCardDBUpdater_20"
    get_last_message_properties(namespace, topic, subscription)

场景2：接收指定会话ID的最后一条消息 #

已知目标会话ID时，直接指定session_id创建接收器：

#!/usr/bin/python3

from azure.servicebus import ServiceBusClient
from azure.identity import DefaultAzureCredential

def get_last_message_properties(namespace_name, topic_name, subscription_name, session_id):
    try:
        # 创建Service Bus客户端
        servicebus_client = ServiceBusClient(
            fully_qualified_namespace=namespace_name,
            credential=DefaultAzureCredential()
        )

        # 获取指定会话的接收器
        with servicebus_client:
            with servicebus_client.get_subscription_session_receiver(
                topic_name=topic_name,
                subscription_name=subscription_name,
                session_id=session_id
            ) as session_receiver:
                # 接收会话中的最后1条消息
                messages = session_receiver.receive(max_message_count=1)
                for message in messages:
                    try:
                        # 访问消息属性
                        properties = message.properties
                        cdId = properties.get("cdId")
                        active = properties.get("active")
                        alarm = properties.get("alarm")
                        print(f"会话 {session_id} 的最后消息属性: cdId={cdId}, active={active}, alarm={alarm}")
                        # 完成消息（若不需要保留消息）
                        message.complete()
                    except Exception as e:
                        print(f"处理消息时出错: {e}")

    except Exception as e:
        print(f"整体执行出错: {e}")

if __name__ == "__main__":
    namespace = "sbdev.servicebus.windows.net"
    topic = "topicsystemlog_main"
    subscription = "SubCardDBUpdater_20"
    target_session_id = "your-target-session-id"  # 替换为实际会话ID
    get_last_message_properties(namespace, topic, subscription, target_session_id)

关键修改说明 #

• 替换原有的get_subscription_receiver为会话接收器方法：accept_next_session或get_subscription_session_receiver
• 使用with语句管理会话接收器生命周期，确保会话正确关闭
• 可选：调用message.complete()标记消息为已处理（若无需保留消息）

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