使用Azure Service Bus Python SDK消费消息时遇消息编号获取异常

问题 #

Azure Service Bus队列中存有约200万条消息，使用以下代码批量消费（每次最多拉取1万条），在成功处理约80万条消息后，触发了Unable to retrieve last received message number异常。

示例代码 #

def consume_messages_from_queue(service_bus_queue: ServiceBusQueue) -> list[dict]:
    """
    return at max 10000 consumed msgs from service bus queue with max wait time as 15 seconds
    """
    with ServiceBusClient.from_connection_string(conn_str=service_bus_queue.conn_str, transport_type=TransportType.AmqpOverWebsocket) as servicebus_client:
        with servicebus_client.get_queue_receiver(queue_name=service_bus_queue.queue_name) as receiver:
            received_msgs = receiver.receive_messages(
                max_message_count=10000, max_wait_time=15)
            received_msgs_json = []
            for msg in received_msgs:
                try:
                    received_msgs_json.append(json.loads(str(msg)))
                    receiver.complete_message(msg)
                except Exception as ex:
                    print(
                        f'Exception in converting service bus message to json: {ex}')
                    receiver.dead_letter_message(message=msg, reason=str(ex))
            return received_msgs_json

异常堆栈信息 #

File ~/cluster-env/clonedenv/lib/python3.10/site-packages/uamqp/receiver.py:229, in MessageReceiver._message_received(self, message)
    221 """Callback run on receipt of every message. If there is
    222 a user-defined callback, this will be called.
    223 Additionally if the client is retrieving messages for a batch
   (...)
    226 :param message: c_uamqp.Message
    227 """
    228 # pylint: disable=protected-access
--> 229 message_number = self._receiver.last_received_message_number()
    230 if self._settle_mode == constants.ReceiverSettleMode.ReceiveAndDelete:
    231     settler = None

File ./src/message_receiver.pyx:67, in uamqp.c_uamqp.cMessageReceiver.last_received_message_number()

File ./src/message_receiver.pyx:70, in uamqp.c_uamqp.cMessageReceiver.last_received_message_number()

File ./src/base.pyx:31, in uamqp.c_uamqp.StructBase._value_error()

ValueError: Operation failed.
Error: Unable to retrieve last received message number.

解决方案 #

针对该异常，可尝试以下修复方式：

• 降低单次批量拉取量：把max_message_count从10000调小（比如改为1000或2000），过大的批量会增加AMQP层处理压力，容易引发元数据（如消息序号）获取失败。
• 循环分批+重建接收器：不要在同一个接收器实例中持续处理数百万条消息，每处理完几批后关闭当前接收器，重新创建新实例。比如每处理5批（5万条）就重启一次接收器，避免长时间连接导致的状态异常。
• 升级依赖库：该异常可能是uamqp库的已知bug，升级到最新版本的azure-servicebus和uamqp可修复部分底层问题，执行命令：

  pip install --upgrade azure-servicebus uamqp
  

• 调整接收模式：若业务允许，使用ReceiveAndDelete模式（创建接收器时指定receive_mode=ServiceBusReceiveMode.RECEIVE_AND_DELETE），这种模式无需跟踪消息序号，能规避该异常，但需确保消息处理逻辑是幂等的，避免丢失风险。
• 优化消息处理逻辑：简化消息解析和完成操作的耗时，避免单条消息处理过久导致接收器连接超时或状态异常。

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