在Azure Functions Python v2中将Azure Service Bus队列消息解码为JSON

解决方法 #

对于ServiceBusReceivedMessage类型的消息，需通过body属性获取原始字节数据，解码为字符串后再解析成JSON对象。之前直接对msg或msg.message调用decode方法失败，是因为这两个对象是封装后的消息实例，并非原始字节流。

修改后的代码示例 #

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

app = func.FunctionApp()

@app.function_name(name="mytimer")
@app.schedule(schedule="0 */30 * * * *", arg_name="mytimer", run_on_startup=True) 
def procesReadings(mytimer: func.TimerRequest) -> None:
    servicebus_client = ServiceBusClient.from_connection_string(conn_str=CONNECTION_STR)

    with servicebus_client:
        receiver = servicebus_client.get_queue_receiver(queue_name=QUEUE_NAME)
        with receiver:
            received_msgs = receiver.peek_messages(max_message_count=5)
            for msg in received_msgs:
                # 获取消息字节内容并解码为UTF-8字符串
                msg_str = msg.body.decode('utf-8')
                # 解析为JSON对象
                try:
                    msg_json = json.loads(msg_str)
                    logging.info(f"解析后的JSON消息: {msg_json}")
                    logging.info(f"消息类型: {type(msg_json)}")
                except json.JSONDecodeError as e:
                    logging.error(f"JSON解析失败: {str(e)}")

补充说明 #

• 若消息体是标准JSON字节流，上述代码可直接完成解析。
• 若消息采用了其他序列化方式（如Avro、Protobuf），需先对应反序列化，再转换为JSON对象。
• 加入异常捕获可避免因消息格式不符合JSON规范导致函数运行报错。

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