Azure Service Bus死信队列如何以Peek模式读取最新消息

问题根源 #

你当前代码用receive_messages结合PeekLock模式，存在两个核心问题：

1. 该方法默认从队列头部（最早的消息）开始读取，无法直接定位到最新插入的消息；
2. 消息被锁定后，若未调用complete()或abandon()，重复调用会返回同一条锁定消息，导致你看到重复的序列号。

要实现仅读取最新消息且不删除/锁定消息，必须使用专门的peek_messages方法，结合消息序列号定位。

解决方案逻辑 #

Service Bus的消息序列号是严格递增的，最新插入的消息拥有最大的序列号。我们可以通过队列运行时属性获取这个最大序列号，再用peek_messages直接读取对应的消息。

修正后的代码 #

from azure.servicebus import ServiceBusClient, ServiceBusSubQueue, TransportType

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

servicebus_client = ServiceBusClient.from_connection_string(
    conn_str=cs,
    retry_total=5, 
    retry_backoff_factor=5, 
    retry_mode="fixed",
    transport_type=TransportType.AmqpOverWebsocket
)

with servicebus_client.get_queue_receiver(
    queue_name=QUEUE_NAME, 
    sub_queue=ServiceBusSubQueue.DEAD_LETTER
) as dlq_receiver:
    # 获取DLQ的运行时属性，提取最新消息的序列号
    queue_runtime = dlq_receiver.get_queue_runtime_properties()
    latest_sequence_number = queue_runtime.last_sequence_number

    if latest_sequence_number is not None:
        # 从最新序列号开始Peek，仅读取1条消息
        peeked_msgs = dlq_receiver.peek_messages(
            from_sequence_number=latest_sequence_number, 
            max_message_count=1
        )
        for msg in peeked_msgs:
            try:
                decoded_value = msg.body.decode('utf-8')
            except UnicodeDecodeError:
                decoded_value = gzip.decompress(msg.body).decode('utf-8')
            print(decoded_value)
            print(f"Body type: {msg.body_type}")
            print(f"Time to live: {msg.time_to_live}")
            print(f"Sequence number: {msg.sequence_number}")
            print(f"Enqueue Sequence number: {msg.enqueued_sequence_number}")
            print(f"Partition Key: {msg.partition_key}")
            print(f"Dead Letter Reason: {msg.dead_letter_reason}")
            print(f"Enqueued time: {msg.enqueued_time_utc}")

关键细节说明 #

• get_queue_runtime_properties()：获取DLQ的实时运行数据，其中last_sequence_number字段就是最新插入消息的唯一标识；
• peek_messages：纯读取模式，不会对消息加锁或删除，完全满足你不修改DLQ消息的需求；
• 指定from_sequence_number=latest_sequence_number：直接定位到最新消息，无需遍历所有旧消息，彻底避免重复读取的问题。

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