使用Python SDK操作ServiceBus时的消息锁过期问题

使用Python SDK操作Azure ServiceBus时遭遇消息锁过期异常：azure.servicebus.exceptions.ServiceBusError: The lock on the message lock has expired。场景为本地运行的死信队列清理工具，需处理多个队列/订阅中积压数月的死信消息。尝试通过AutoLockRenewer续期锁，但异常仍会出现。每次工具崩溃重启后，可继续处理之前失败的消息，但会在其他队列/订阅再次因锁过期崩溃，多次重启后能清理更多队列。

代码片段：

renewer = AutoLockRenewer()
with ServiceBusClient.from_connection_string(shared_access_key["primaryConnectionString"]) as client:
for queue in queues:
    if queue["countDetails"]["deadLetterMessageCount"] > 0:
        with client.get_queue_receiver(queue_name=queue["name"], sub_queue="deadletter") as receiver:
            while len(receiver.receive_messages(max_wait_time=60)) > 0:
                messages = receiver.receive_messages(max_message_count=50, max_wait_time=60)
                for message in messages:
                    renewer.register(receiver, message, max_lock_renewal_duration=300)
                    receiver.complete_message(message)

问题分析与修复 #

1. 重复接收消息浪费锁资源 #

代码中while循环内首次调用receiver.receive_messages()后未处理这批消息，直接丢弃。这不仅浪费了消息锁的有效期，还会导致这批消息因锁过期回流队列，增加后续重复处理的开销。

2. AutoLockRenewer注册时机过晚 #

在调用complete_message()前才注册续期，此时消息的锁可能已接近过期甚至失效。正确做法是接收到消息后立即注册续期，确保整个处理周期内锁能被自动续期。

3. 优化后的代码 #

from azure.servicebus import ServiceBusClient, AutoLockRenewer

renewer = AutoLockRenewer()
with ServiceBusClient.from_connection_string(shared_access_key["primaryConnectionString"]) as client:
    for queue in queues:
        dead_letter_count = queue["countDetails"]["deadLetterMessageCount"]
        if dead_letter_count <= 0:
            continue
            
        with client.get_queue_receiver(queue_name=queue["name"], sub_queue="deadletter") as receiver:
            # 调整循环逻辑，避免重复接收消息
            while True:
                messages = receiver.receive_messages(max_message_count=50, max_wait_time=60)
                if not messages:
                    break  # 无消息时退出循环
                    
                # 先为所有消息注册锁续期
                for message in messages:
                    renewer.register(receiver, message, max_lock_renewal_duration=300)
                
                # 再批量完成消息
                for message in messages:
                    receiver.complete_message(message)

额外优化建议 #

• 若单条消息处理耗时较长，可在创建接收器时通过max_lock_duration设置更长的初始锁时长（最大值为5分钟）：

  client.get_queue_receiver(
      queue_name=queue["name"], 
      sub_queue="deadletter",
      max_lock_duration=300  # 单位：秒，对应5分钟
  )
  

• 添加异常捕获逻辑，避免工具因单条消息锁过期直接崩溃：

  from azure.servicebus.exceptions import ServiceBusError
  
  # ... 其他代码 ...
  for message in messages:
      renewer.register(receiver, message, max_lock_renewal_duration=300)
      try:
          receiver.complete_message(message)
      except ServiceBusError as e:
          if "lock has expired" in str(e):
              print(f"消息锁过期，跳过消息ID: {message.message_id}")
          else:
              # 其他异常重新抛出
              raise
  

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