MQTT订阅端JSONDecodeError：如何仅在收到有效消息时解码？

问题描述 #

我有两个Python脚本，分别运行在树莓派（RPI）和PC上，尝试通过Mosquitto MQTT将树莓派的指标以JSON格式发送到PC。执行订阅端脚本后，连接成功建立，但出现JSONDecodeError: Expecting value: line 1 column 1 (char 0)错误。推测是因为发布端尚未发送消息时，msg.payload为空，订阅端仍尝试解码JSON。需要确保仅在发布端发送有效消息时才执行JSON解码操作。

问题分析 #

出现该错误的常见原因：

• 收到空消息 payload（比如订阅后Broker推送的空留存消息，或其他设备误发的空消息）
• 收到的消息并非合法JSON格式
• 额外注意：你的发布端发送的是字典结构的JSON，但订阅端尝试用索引json_data[0]访问，即使解码成功也会触发KeyError，这个问题也需要同步修正。

解决方案 #

方法1：先校验payload非空，再执行解码 #

在解码前先判断消息内容是否为空，避免对空数据执行JSON解析：

def on_message(client, userdata, msg):
    payload_str = msg.payload.decode("utf-8")
    # 过滤空消息
    if not payload_str.strip():
        print("收到空消息，跳过处理")
        return
    try:
        json_data = json.loads(payload_str)
    except json.JSONDecodeError as e:
        print(f"JSON解码失败: {e}")
        return
    # 改用键名访问字典内容，替代原索引方式
    print("Variable x:", json_data.get("x"))
    print("Variable y:", json_data.get("y"))
    print("Variable z:", json_data.get("z"))
    print("Variable a:", json_data.get("a"))
    print("Variable b:", json_data.get("b"))
    print("Variable c:", json_data.get("c"))
    print("Variable d:", json_data.get("d"))

方法2：用异常捕获处理所有解码失败场景 #

直接用try-except包裹解码逻辑，不管是空消息还是非法JSON，都能捕获并处理：

def on_message(client, userdata, msg):
    try:
        payload_str = msg.payload.decode("utf-8")
        json_data = json.loads(payload_str)
    except (UnicodeDecodeError, json.JSONDecodeError) as e:
        print(f"消息无效，无法解码: {e}")
        return
    # 使用get方法避免键不存在时触发报错
    print("Variable x:", json_data.get("x"))
    print("Variable y:", json_data.get("y"))
    print("Variable z:", json_data.get("z"))
    print("Variable a:", json_data.get("a"))
    print("Variable b:", json_data.get("b"))
    print("Variable c:", json_data.get("c"))
    print("Variable d:", json_data.get("d"))

额外优化建议 #

• 给MQTT主题设置更具体的前缀（比如rpi/metrics），避免收到无关设备的消息
• 发布端添加消息质量等级（QoS），提升消息可靠性：

  publish.single("rpi/metrics", json_data, hostname="test.mosquitto.org", qos=1)
  

• 订阅端同步设置对应QoS，保持两端一致：

  client.subscribe("rpi/metrics", qos=1)
  

修正后的完整订阅端脚本 #

import paho.mqtt.client as mqtt
import json

def on_connect(client, userdata, flags, rc):
    print("Connected with result code "+str(rc))
    client.subscribe("test", qos=1)

def on_message(client, userdata, msg):
    payload_str = msg.payload.decode("utf-8")
    if not payload_str.strip():
        print("收到空消息，跳过处理")
        return
    try:
        json_data = json.loads(payload_str)
    except json.JSONDecodeError as e:
        print(f"JSON解码失败: {e}")
        return
    # 使用键名访问字典内容，避免索引错误
    print("Variable x:", json_data.get("x"))
    print("Variable y:", json_data.get("y"))
    print("Variable z:", json_data.get("z"))
    print("Variable a:", json_data.get("a"))
    print("Variable b:", json_data.get("b"))
    print("Variable c:", json_data.get("c"))
    print("Variable d:", json_data.get("d"))

client = mqtt.Client()
client.on_connect = on_connect
client.on_message = on_message
client.connect("test.mosquitto.org", 1883, 60)
client.loop_forever()

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