如何在Python的Service Bus Trigger中延迟处理输入消息？

问题核心 #

你遇到的问题是Azure Functions的Service Bus触发器传入的func.ServiceBusMessage与Azure Service Bus SDK的ServiceBusReceivedMessage类型不兼容，且触发器已锁定消息，单独创建的接收器无法操作该消息。

解决方案 #

方法1：使用触发器消息对象的内置延迟方法（推荐） #

从Azure Functions Python Service Bus扩展v1.11.0及以上版本开始，func.ServiceBusMessage已内置defer()方法，无需额外创建接收器即可直接延迟消息。该方法会复用触发器持有的消息锁和连接，避免类型不匹配与锁冲突问题。

示例代码：

import azure.functions as func
import logging

def main(msg: func.ServiceBusMessage):
    logging.info(f"收到消息：{msg.message_id=}, {msg.sequence_number=}")

    # 直接调用内置的延迟方法
    msg.defer()

方法2：通过消息序号延迟（旧版扩展兼容方案） #

如果你使用的是旧版扩展，无法使用内置defer()方法，可以通过消息的sequence_number，结合同一会话的接收器来延迟消息。必须确保接收器与触发器使用同一个会话ID，否则无法访问已被锁定的消息。

示例代码：

import azure.functions as func
import logging
from azure.servicebus import ServiceBusClient

def main(msg: func.ServiceBusMessage):
    logging.info(f"收到消息：{msg.message_id=}, {msg.sequence_number=}")

    # 获取当前消息所属的会话ID
    session_id = msg.session_id

    # 初始化Service Bus客户端
    service_bus_client = ServiceBusClient.from_connection_string(conn_str="你的连接字符串")
    
    # 创建与触发器同一会话的接收器
    with service_bus_client.get_queue_receiver(
        queue_name="test-queue",
        session_id=session_id,
        receive_mode="peek_lock"
    ) as receiver:
        # 通过消息序号执行延迟操作
        receiver.defer_message(sequence_number=msg.sequence_number)

为什么你的原有代码无法工作 #

• 你创建的接收器使用了NEXT_AVAILABLE_SESSION，可能获取到的不是当前消息所在的会话，因此无法访问已被锁定的消息。
• func.ServiceBusMessage是Azure Functions对原始消息的封装类，与SDK的ServiceBusReceivedMessage并非同一类型，无法直接传入SDK的defer_message方法。

内容的提问来源于stack exchange，提问作者O.N.