Python读取Azure ServiceBus消息时遇UnicodeDecodeError求助

问题原因 #

错误'utf-8' codec can't decode byte 0xd8 in position 172: invalid continuation byte是因为部分消息并非标准UTF-8编码：0xd8是UTF-16编码中的高代理字符，单独出现时不符合UTF-8的编码规则，导致解码失败。可能是发送端采用了UTF-16编码，或是消息传输中出现损坏，甚至包含了非文本的二进制数据。

解决方案 #

由于你无法控制消息内容，只能在接收端做健壮性处理，以下是几种可行的修改方式：

1. 优先尝试UTF-16解码，失败则降级处理 #

修改官方脚本中解码消息体的代码，先尝试UTF-16（对应0xd8字节的常见场景），如果失败再用UTF-8替换错误字节：

from azure.servicebus import ServiceBusClient

# 其他初始化代码保持不变
with servicebus_client:
    receiver = servicebus_client.get_queue_receiver(queue_name=QUEUE_NAME)
    with receiver:
        for msg in receiver:
            try:
                # 先尝试UTF-16解码
                msg_body = msg.body.decode('utf-16')
            except UnicodeDecodeError:
                # 失败则用UTF-8替换错误字节
                msg_body = msg.body.decode('utf-8', errors='replace')
            # 后续处理JSON和保存到T-SQL的代码
            receiver.complete_message(msg)

2. 自动检测消息编码 #

用chardet库自动识别消息的编码类型，提升兼容性：
首先安装依赖：
pip install chardet

然后修改解码逻辑：

from azure.servicebus import ServiceBusClient
import chardet

# 其他初始化代码保持不变
with servicebus_client:
    receiver = servicebus_client.get_queue_receiver(queue_name=QUEUE_NAME)
    with receiver:
        for msg in receiver:
            raw_data = msg.body
            # 检测编码
            detect_result = chardet.detect(raw_data)
            target_encoding = detect_result['encoding'] or 'utf-8'
            try:
                msg_body = raw_data.decode(target_encoding)
            except UnicodeDecodeError:
                # 最终降级到UTF-8替换错误
                msg_body = raw_data.decode('utf-8', errors='replace')
            # 后续处理JSON和保存到T-SQL的代码
            receiver.complete_message(msg)

3. 强制忽略错误，保证脚本不崩溃 #

如果不需要保留错误字节的内容，直接用errors='ignore'或errors='replace'跳过错误，确保脚本持续运行：

# 用?替换错误字节
msg_body = msg.body.decode('utf-8', errors='replace')
# 或者直接丢弃错误字节
# msg_body = msg.body.decode('utf-8', errors='ignore')

额外建议 #

可以在解码失败时记录消息ID和错误信息，方便后续排查（比如联系发送端确认编码方式）：

try:
    msg_body = msg.body.decode('utf-8')
except UnicodeDecodeError as e:
    print(f"消息ID {msg.message_id} 解码失败：{str(e)}")
    msg_body = msg.body.decode('utf-8', errors='replace')

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