【IoT进阶】MQTT Protobuf 二进制通信实战从 Python 双向控制到 Wireshark 抓包可视化全解析摘要在物联网IoT、车联网与工业场景中相较于冗长的 JSON 文本MQTT Google Protobuf是兼顾高吞吐、极低功耗与强类型扩展的事实工业标准。本文以“智能宠物喝水器”为实战案例手把手带你完成协议设计、Python 双端通信并通过Wireshark Lua 插件实现 Protobuf 二进制 Payload 的无缝抓包与字段可视化。一、 为什么选择 MQTT Protobuf在低功耗传感器如 NB-IoT 水表、电池供电设备或大规模车联网中JSON 文本键名重复占用大量字节序列化/反序列化消耗 CPU流量与电量成本高。Protobuf 二进制采用 Varint 变长整型压缩与强类型契约体积比 JSON 缩减80%~90%且具备完美的向前/向后兼容性。二、 协议设计与定义我们设计一套智能喝水器的物联网通信协议包含上行遥测状态Device - Cloud/App与下行控制指令App - Device。1. 协议数据结构设计① 上行状态报文Uplink TelemetryTopic:my_pet_feeder_proto/001/status0 1 2 3 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 -------------------------------- | Device ID (2 Bytes) | Water Level(1)| Status (1) | -------------------------------- | Timestamp (4 Bytes) | -------------------------------- | Battery Voltage (Float 4 Bytes) | --------------------------------Device ID(uint32): 设备编号Water Level(uint32): 水位0 ~ 100(%)Status(Enum):0NORMAL,1LOW_WATER,2EMPTYTimestamp(uint64): Unix 秒级时间戳Battery Voltage(float): 电池实时电压 (V)② 下行控制报文Downlink CommandTopic:my_pet_feeder_proto/001/command0 1 2 3 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 -------------------------------- | Device ID (2 Bytes) | Action (1) | Target Val (1)| --------------------------------Device ID(uint32): 目标设备编号Action(Enum):0REFILL (远程灌装加水),1BEEP (蜂鸣寻宠)Target Val(uint32): 加水目标百分比如 1002. Protobuf 契约文件feeder.proto新建feeder.protosyntax proto3; package iot.pet; // 1. 上行状态报文 (设备 - App) message FeederStatus { uint32 device_id 1; // 设备ID uint32 water_level 2; // 水位百分比 (0-100) enum StatusEnum { NORMAL 0; // 正常 LOW_WATER 1; // 低水位预警 EMPTY 2; // 缺水 } StatusEnum status 3; // 设备状态枚举 uint64 timestamp 4; // 时间戳 float battery_voltage 5; // 电池电压 } // 2. 下行控制报文 (App - 设备) message FeederCommand { uint32 device_id 1; // 目标设备ID enum ActionEnum { REFILL 0; // 灌装加水 BEEP 1; // 蜂鸣寻宠 } ActionEnum action 2; // 指令动作 uint32 target_val 3; // 目标参数值 }编译生成 Python 类库pipinstallprotobuf grpcio-tools paho-mqtt python-mgrpc_tools.protoc -I.--python_out. feeder.proto生成feeder_pb2.py备用。三、 Python 双端通信代码实现1. 设备端硬件模拟feeder_device.pyimportpaho.mqtt.clientasmqttimporttimeimportthreadingimportfeeder_pb2 TOPIC_STATUSmy_pet_feeder_proto/001/statusTOPIC_COMMANDmy_pet_feeder_proto/001/commandDEVICE_ID1water_level100lockthreading.Lock()defon_message(client,userdata,msg):globalwater_leveltry:cmdfeeder_pb2.FeederCommand()cmd.ParseFromString(msg.payload)# Protobuf 反序列化print(f\n[收到下行指令] 动作:{feeder_pb2.FeederCommand.ActionEnum.Name(cmd.action)}, 目标值:{cmd.target_val}%)ifcmd.device_idDEVICE_IDandcmd.actionfeeder_pb2.FeederCommand.REFILL:withlock:water_levelcmd.target_valprint(f [水泵启动] 灌装加水完成当前水位:{water_level}% \n)exceptExceptionase:print(指令反序列化失败:,e)defon_connect(client,userdata,flags,rc,propertiesNone):print([喝水器] 连接 Broker 成功正在监听控制指令...)client.subscribe(TOPIC_COMMAND,qos1)clientmqtt.Client(mqtt.CallbackAPIVersion.VERSION2,client_idproto_feeder_001)client.on_connecton_connect client.on_messageon_message client.connect(broker.emqx.io,1883,keepalive60)client.loop_start()try:print([喝水器] 启动 Protobuf 上报循环...)whileTrue:withlock:ifwater_level0:water_level-15ifwater_level0:water_level0cur_levelwater_level# 构造 Protobuf 实体statusfeeder_pb2.FeederStatus()status.device_idDEVICE_ID status.water_levelcur_level status.timestampint(time.time())status.battery_voltage3.95ifcur_level30:status.statusfeeder_pb2.FeederStatus.NORMALelifcur_level0:status.statusfeeder_pb2.FeederStatus.LOW_WATERelse:status.statusfeeder_pb2.FeederStatus.EMPTY# 序列化为二进制流发布binary_payloadstatus.SerializeToString()client.publish(TOPIC_STATUS,binary_payload,qos0)print(f[上报] HEX[{binary_payload.hex( )}] ({len(binary_payload)}B) | 水位{cur_level}% | 电压{status.battery_voltage:.2f}V)time.sleep(2)exceptKeyboardInterrupt:client.loop_stop()client.disconnect()2. 控制端主人 Appowner_app.pyimportpaho.mqtt.clientasmqttimporttimeimportfeeder_pb2 TOPIC_STATUSmy_pet_feeder_proto/001/statusTOPIC_COMMANDmy_pet_feeder_proto/001/commanddefon_message(client,userdata,msg):try:statusfeeder_pb2.FeederStatus()status.ParseFromString(msg.payload)levelstatus.water_level status_textfeeder_pb2.FeederStatus.StatusEnum.Name(status.status)bar█*(level//10)-*(10-level//10)print(f\r[主人 App 看板] 设备#{status.device_id}水位:[{bar}]{level:3d}% | 状态:{status_text:9}| 电池:{status.battery_voltage:.2f}V,end,flushTrue)iflevel0:print(\n⚠️ 小狗没水了输入 1 立即加水: ,end,flushTrue)exceptExceptionase:print(\n解析失败:,e)defon_connect(client,userdata,flags,rc,propertiesNone):print([主人 App] 连接成功正在订阅 Protobuf 数据流...)client.subscribe(TOPIC_STATUS,qos0)clientmqtt.Client(mqtt.CallbackAPIVersion.VERSION2,client_idproto_owner_app)client.on_connecton_connect client.on_messageon_message client.connect(broker.emqx.io,1883,keepalive60)client.loop_start()print(\n 智能喝水器控制台 (输入 1 远程加水q 退出) )try:whileTrue:cmdinput()ifcmd.strip()in[1,refill]:cmd_pbfeeder_pb2.FeederCommand()cmd_pb.device_id1cmd_pb.actionfeeder_pb2.FeederCommand.REFILL cmd_pb.target_val100client.publish(TOPIC_COMMAND,cmd_pb.SerializeToString(),qos1)print(f\n--- 已发送 Protobuf 灌装指令)elifcmd.strip().lower()q:breakexceptKeyboardInterrupt:passclient.loop_stop()client.disconnect()四、 Wireshark 抓包与 Protobuf 可视化实战这是整个方案最核心、最惊艳的环节让 Wireshark 直接将二进制 Payload 还原为结构化字段树。1. 编写 Wireshark Lua 桥接插件新建文件mqtt_feeder_protobuf.lua-- MQTT payload decoders for feeder.protolocalprotobufDissector.get(protobuf)-- 注册状态报文解析器localfeeder_statusProto(feeder_status_pb,Pet Feeder Status (Protobuf))functionfeeder_status.dissector(tvb,pinfo,tree)pinfo.private[pb_msg_type]message,iot.pet.FeederStatuspcall(Dissector.call,protobuf,tvb,pinfo,tree)end-- 注册控制指令解析器localfeeder_commandProto(feeder_command_pb,Pet Feeder Command (Protobuf))functionfeeder_command.dissector(tvb,pinfo,tree)pinfo.private[pb_msg_type]message,iot.pet.FeederCommandpcall(Dissector.call,protobuf,tvb,pinfo,tree)end2. Wireshark 详细配置步骤步骤 ①配置.proto搜索路径将feeder.proto放置在一个固定目录例如C:\Users\xing\PycharmProjects\testMQTT\protobuf-mqtt。打开 Wireshark -编辑 (Edit)-首选项 (Preferences)- 展开Protocols-Protobuf。在Protobuf search paths点击Edit添加上述目录并勾选启用。步骤 ②安装 Lua 插件将mqtt_feeder_protobuf.lua复制到 Wireshark 的插件目录Windows 路径C:\Users\用户名\AppData\Roaming\Wireshark\plugins\。步骤 ③绑定 MQTT Topic 与 Payload 解码器打开 Wireshark -编辑 (Edit)-首选项 (Preferences)-Protocols-MQTT。找到Message Decoding表格点击Edit新增映射Topic:my_pet_feeder_proto/001/statusPayload dissector:feeder_status_pbDecoding:none重启 Wireshark 或按下快捷键Ctrl Shift L重新加载脚本。五、 抓包验证效果过滤条件输入tcp.port 1883点击任意一条 Publish Message可以看到 Wireshark 完美的解析效果 MQ Telemetry Transport Protocol, Publish Message Topic: my_pet_feeder_proto/001/status Message: 0801180220d6b5e6d3062dcdcc7c40 [Message decoded as: feeder_status_pb] v Protocol Buffers: iot.pet.FeederStatus v Message: iot.pet.FeederStatus Field(1): device_id 1 (uint32) Field(3): status EMPTY(2) (enum) Field(4): timestamp 1786354390 (uint64) Field(5): battery_voltage 3.950000 (float)原本一长串十六进制字节0801180220d6...在 Wireshark 中被清晰地拆解出了设备 ID、状态枚举、时间戳以及精准的浮点数3.95V 电池电压六、 总结通过本方案网络层利用 MQTT 提供了可靠的 QoS 消息投递与 Pub/Sub 解耦。数据层利用 Protobuf 实现了极度紧凑的二进制序列化极大降低了 IoT 设备的功耗与流量成本。运维/诊断层利用 Wireshark Lua 插件打通了从底层抓包到业务字段的可视化使二进制排错如同看 JSON 一样直观。该架构正是Sparkplug B等现代工业物联网与车联网核心通信方案的标准实践方式。