当前位置：首页 > news >正文

[Spring Boot]Protobuf解析MQTT消息体

news 2025/7/1 21:42:11

简述

本文主要针对在MQTT场景下，使用Protobuf协议解析MQTT的消息体

Protobuf下载

官方下载

https://github.com/protocolbuffers/protobuf/releases

网盘下载

链接：https://pan.baidu.com/s/1Uz7CZuOSwa8VCDl-6r2xzw?pwd=anan 
提取码：anan

本文项目代码

java_fir_mqtt-proto
https://gitee.com/dong-puen/fir_public

配置protoc到Windows系统

解压文件，配置环境变量：

PROTOCBUF_HOME

Path中添加：

%PROTOCBUF_HOME%\bin

命令行验证

protoc

配置protoc到Ubuntu 22.04.4

安装

apt-get install libprotobuf-dev protobuf-compiler

验证

protoc

Java集成

MQTT依赖

        <!--        mqtt连接依赖--><dependency><groupId>com.hivemq</groupId><artifactId>hivemq-mqtt-client</artifactId><version>1.3.3</version></dependency>

编辑proto文件

syntax = "proto3";// 是否开启多文件
option java_multiple_files = true;
// 包命
option java_package = "com.fir.proto.protobuf.info";
// Java外部类名
option java_outer_classname = "Info";message Info {// 字符串string name = 1;// 整数类型int32 int32_value = 3;int64 int64_value = 4;uint32 uint32_value = 5;uint64 uint64_value = 6;sint32 sint32_value = 7;sint64 sint64_value = 8;fixed32 fixed32_value = 9;fixed64 fixed64_value = 10;sfixed32 sfixed32_value = 11;sfixed64 sfixed64_value = 12;// 布尔类型bool bool_value = 13;// 浮点类型float float_value = 14;double double_value = 15;// 字节类型bytes bytes_value = 16;// 枚举类型enum EnumType {DEFAULT = 0;TYPE1 = 1;TYPE2 = 2;
}EnumType enum_value = 17;// 映射类型map<string, int32> map_value = 18;// 内嵌消息类型message NestedMessage {string nested_field = 1;
}NestedMessage nested_message = 19;// 重复字段repeated string repeated_strings = 20;
}

命令行生成Java文件

protoc --java_out=C:\proto test.proto

（ps:虽然有地方推荐使用idea生成，但是实测有时候会卡住，反而命令行生成速度更快，这里推荐使用命令行。）

复制生成的3个文件

发送数据与解析数据

发送数据

初始化数据

    @GetMapping("/send")public Object pageData() {// 构建MesInfo对象MesInfo mesInfo = MesInfo.newBuilder().setName("测试数据").build();mqttClient.send(mqttClient.TEST_MSG, mesInfo.toByteArray());return true;}

也可以初始化所有数据对象

    @GetMapping("/send")public Object pageData() {// 初始化MesInfo对象MesInfo.Builder mesInfoBuilder = MesInfo.newBuilder();// 设置字符串字段mesInfoBuilder.setName("测试数据");// 设置整数类型字段mesInfoBuilder.setInt32Value(32);mesInfoBuilder.setInt64Value(64L);mesInfoBuilder.setUint32Value(32);mesInfoBuilder.setUint64Value(64L);mesInfoBuilder.setSint32Value(32);mesInfoBuilder.setSint64Value(64L);mesInfoBuilder.setFixed32Value(32);mesInfoBuilder.setFixed64Value(64L);mesInfoBuilder.setSfixed32Value(32);mesInfoBuilder.setSfixed64Value(64L);// 设置布尔类型字段mesInfoBuilder.setBoolValue(true);// 设置浮点类型字段mesInfoBuilder.setFloatValue(3.14f);mesInfoBuilder.setDoubleValue(3.141592653589793);// 设置字节类型字段mesInfoBuilder.setBytesValue(ByteString.copyFromUtf8("字节字符串"));// 设置枚举类型字段mesInfoBuilder.setEnumValue(MesInfo.EnumType.TYPE1);// 设置映射类型字段mesInfoBuilder.putMapValue("key1", 100);mesInfoBuilder.putMapValue("key2", 200);// 设置内嵌消息类型字段MesInfo.NestedMessage.Builder nestedMessageBuilder = MesInfo.NestedMessage.newBuilder();nestedMessageBuilder.setNestedField("内嵌字段值");mesInfoBuilder.setNestedMessage(nestedMessageBuilder.build());// 设置重复字段mesInfoBuilder.addRepeatedStrings("字符串1");mesInfoBuilder.addRepeatedStrings("字符串2");// 构建MesInfo对象MesInfo mesInfo = mesInfoBuilder.build();mqttClient.send(mqttClient.TEST_MSG, mesInfo.toByteArray());return true;}

解析数据

    @Bean(name = "mqttBackBean")public Consumer<Mqtt3Publish> mqttBackBean() {SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");String datetime = sdf.format(new Date());log.info("[MQTT]:" + datetime);Consumer<Mqtt3Publish> callback = mqtt3Publish -> {String topic;topic = mqtt3Publish.getTopic().toString();byte[] mqttData = mqtt3Publish.getPayloadAsBytes();log.info("[MQTT] 主题{} 消息体 {}}", topic, Arrays.toString(mqttData));try {MesInfo mesInfo =  MesInfo.parseFrom(mqttData);System.out.println("mesInfo============================================================");System.out.println(mesInfo.toString());System.out.println(mesInfo);System.out.println("mesInfo============================================================");// 打印所有字段for (Map.Entry<Descriptors.FieldDescriptor, Object> entry : mesInfo.getAllFields().entrySet()) {Descriptors.FieldDescriptor field = entry.getKey();Object value = entry.getValue();System.out.println(field.getName() + ": " + value);}} catch (Exception e) {e.printStackTrace();// pass}};log.info("[MQTT] 处理结束:" + datetime);return callback;}